working on sensors and rules
This commit is contained in:
@@ -14,9 +14,11 @@ pub fn create_cors_layer(allowed_origins: Vec<String>) -> CorsLayer {
|
||||
// Default development origins
|
||||
vec![
|
||||
"http://localhost:3000".to_string(),
|
||||
"http://localhost:3001".to_string(),
|
||||
"http://localhost:5173".to_string(),
|
||||
"http://localhost:8080".to_string(),
|
||||
"http://127.0.0.1:3000".to_string(),
|
||||
"http://127.0.0.1:3001".to_string(),
|
||||
"http://127.0.0.1:5173".to_string(),
|
||||
"http://127.0.0.1:8080".to_string(),
|
||||
]
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
//! Parameter validation module
|
||||
//!
|
||||
//! Validates trigger and action parameters against their declared JSON schemas.
|
||||
//! Template-aware: values containing `{{ }}` template expressions are replaced
|
||||
//! with schema-appropriate placeholders before validation, so template expressions
|
||||
//! pass type checks while literal values are still validated normally.
|
||||
|
||||
use attune_common::models::{action::Action, trigger::Trigger};
|
||||
use jsonschema::Validator;
|
||||
@@ -8,15 +11,167 @@ use serde_json::Value;
|
||||
|
||||
use crate::middleware::ApiError;
|
||||
|
||||
/// Validate trigger parameters against the trigger's parameter schema
|
||||
/// Check if a JSON value is (or contains) a template expression.
|
||||
fn is_template_expression(value: &Value) -> bool {
|
||||
match value {
|
||||
Value::String(s) => s.contains("{{") && s.contains("}}"),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Given a JSON Schema property definition, produce a placeholder value that
|
||||
/// satisfies the schema's type constraint. This is used to replace template
|
||||
/// expressions so that JSON Schema validation passes for the remaining
|
||||
/// (non-template) parts of the parameters.
|
||||
fn placeholder_for_schema(property_schema: &Value) -> Value {
|
||||
// Handle anyOf / oneOf by picking the first variant
|
||||
if let Some(any_of) = property_schema.get("anyOf").and_then(|v| v.as_array()) {
|
||||
if let Some(first) = any_of.first() {
|
||||
return placeholder_for_schema(first);
|
||||
}
|
||||
}
|
||||
if let Some(one_of) = property_schema.get("oneOf").and_then(|v| v.as_array()) {
|
||||
if let Some(first) = one_of.first() {
|
||||
return placeholder_for_schema(first);
|
||||
}
|
||||
}
|
||||
|
||||
let type_value = property_schema.get("type").and_then(|t| t.as_str());
|
||||
|
||||
match type_value {
|
||||
Some("integer") => {
|
||||
// Use minimum if set, else default if set, else 0
|
||||
if let Some(default) = property_schema.get("default") {
|
||||
return default.clone();
|
||||
}
|
||||
if let Some(min) = property_schema.get("minimum").and_then(|v| v.as_i64()) {
|
||||
return Value::Number(min.into());
|
||||
}
|
||||
Value::Number(0.into())
|
||||
}
|
||||
Some("number") => {
|
||||
if let Some(default) = property_schema.get("default") {
|
||||
return default.clone();
|
||||
}
|
||||
if let Some(min) = property_schema.get("minimum").and_then(|v| v.as_f64()) {
|
||||
return serde_json::Number::from_f64(min)
|
||||
.map(Value::Number)
|
||||
.unwrap_or(Value::Number(0.into()));
|
||||
}
|
||||
serde_json::Number::from_f64(0.0)
|
||||
.map(Value::Number)
|
||||
.unwrap_or(Value::Number(0.into()))
|
||||
}
|
||||
Some("boolean") => {
|
||||
if let Some(default) = property_schema.get("default") {
|
||||
return default.clone();
|
||||
}
|
||||
Value::Bool(true)
|
||||
}
|
||||
Some("array") => {
|
||||
if let Some(default) = property_schema.get("default") {
|
||||
return default.clone();
|
||||
}
|
||||
Value::Array(vec![])
|
||||
}
|
||||
Some("object") => {
|
||||
if let Some(default) = property_schema.get("default") {
|
||||
return default.clone();
|
||||
}
|
||||
Value::Object(serde_json::Map::new())
|
||||
}
|
||||
Some("string") | None => {
|
||||
// For enum fields, use the first valid value so enum validation passes
|
||||
if let Some(enum_values) = property_schema.get("enum").and_then(|v| v.as_array()) {
|
||||
if let Some(first) = enum_values.first() {
|
||||
return first.clone();
|
||||
}
|
||||
}
|
||||
if let Some(default) = property_schema.get("default") {
|
||||
return default.clone();
|
||||
}
|
||||
Value::String("__template_placeholder__".to_string())
|
||||
}
|
||||
Some(_) => Value::Null,
|
||||
}
|
||||
}
|
||||
|
||||
/// Walk a parameters object and replace any template expression values with
|
||||
/// schema-appropriate placeholders. Only replaces leaf values that match
|
||||
/// `{{ ... }}`; non-template values are left untouched for normal validation.
|
||||
///
|
||||
/// `schema` should be the full JSON Schema object (with `properties`, `type`, etc).
|
||||
fn replace_templates_with_placeholders(params: &Value, schema: &Value) -> Value {
|
||||
match params {
|
||||
Value::Object(map) => {
|
||||
let properties = schema.get("properties").and_then(|p| p.as_object());
|
||||
|
||||
let mut result = serde_json::Map::new();
|
||||
for (key, value) in map {
|
||||
let prop_schema = properties.and_then(|p| p.get(key));
|
||||
|
||||
if is_template_expression(value) {
|
||||
// Replace with a type-appropriate placeholder
|
||||
if let Some(ps) = prop_schema {
|
||||
result.insert(key.clone(), placeholder_for_schema(ps));
|
||||
} else {
|
||||
// No schema for this property — keep as string placeholder
|
||||
result.insert(
|
||||
key.clone(),
|
||||
Value::String("__template_placeholder__".to_string()),
|
||||
);
|
||||
}
|
||||
} else if value.is_object() {
|
||||
// Recurse into nested objects
|
||||
let empty_schema = Value::Object(serde_json::Map::new());
|
||||
let nested_schema = prop_schema.unwrap_or(&empty_schema);
|
||||
result.insert(
|
||||
key.clone(),
|
||||
replace_templates_with_placeholders(value, nested_schema),
|
||||
);
|
||||
} else if value.is_array() {
|
||||
// Recurse into arrays — check each element
|
||||
if let Some(arr) = value.as_array() {
|
||||
let empty_items_schema = Value::Object(serde_json::Map::new());
|
||||
let item_schema = prop_schema
|
||||
.and_then(|ps| ps.get("items"))
|
||||
.unwrap_or(&empty_items_schema);
|
||||
let new_arr: Vec<Value> = arr
|
||||
.iter()
|
||||
.map(|item| {
|
||||
if is_template_expression(item) {
|
||||
placeholder_for_schema(item_schema)
|
||||
} else if item.is_object() || item.is_array() {
|
||||
replace_templates_with_placeholders(item, item_schema)
|
||||
} else {
|
||||
item.clone()
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
result.insert(key.clone(), Value::Array(new_arr));
|
||||
} else {
|
||||
result.insert(key.clone(), value.clone());
|
||||
}
|
||||
} else {
|
||||
result.insert(key.clone(), value.clone());
|
||||
}
|
||||
}
|
||||
Value::Object(result)
|
||||
}
|
||||
other => other.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate trigger parameters against the trigger's parameter schema.
|
||||
/// Template expressions (`{{ ... }}`) are accepted for any field type.
|
||||
pub fn validate_trigger_params(trigger: &Trigger, params: &Value) -> Result<(), ApiError> {
|
||||
// If no schema is defined, accept any parameters
|
||||
let Some(schema) = &trigger.param_schema else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
// If parameters are empty object and schema exists, validate against schema
|
||||
// (schema might allow empty object or have defaults)
|
||||
// Replace template expressions with schema-appropriate placeholders
|
||||
let sanitized = replace_templates_with_placeholders(params, schema);
|
||||
|
||||
// Compile the JSON schema
|
||||
let compiled_schema = Validator::new(schema).map_err(|e| {
|
||||
@@ -26,9 +181,9 @@ pub fn validate_trigger_params(trigger: &Trigger, params: &Value) -> Result<(),
|
||||
))
|
||||
})?;
|
||||
|
||||
// Validate the parameters
|
||||
// Validate the sanitized parameters
|
||||
let errors: Vec<String> = compiled_schema
|
||||
.iter_errors(params)
|
||||
.iter_errors(&sanitized)
|
||||
.map(|e| {
|
||||
let path = e.instance_path().to_string();
|
||||
if path.is_empty() {
|
||||
@@ -50,13 +205,17 @@ pub fn validate_trigger_params(trigger: &Trigger, params: &Value) -> Result<(),
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Validate action parameters against the action's parameter schema
|
||||
/// Validate action parameters against the action's parameter schema.
|
||||
/// Template expressions (`{{ ... }}`) are accepted for any field type.
|
||||
pub fn validate_action_params(action: &Action, params: &Value) -> Result<(), ApiError> {
|
||||
// If no schema is defined, accept any parameters
|
||||
let Some(schema) = &action.param_schema else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
// Replace template expressions with schema-appropriate placeholders
|
||||
let sanitized = replace_templates_with_placeholders(params, schema);
|
||||
|
||||
// Compile the JSON schema
|
||||
let compiled_schema = Validator::new(schema).map_err(|e| {
|
||||
ApiError::InternalServerError(format!(
|
||||
@@ -65,9 +224,9 @@ pub fn validate_action_params(action: &Action, params: &Value) -> Result<(), Api
|
||||
))
|
||||
})?;
|
||||
|
||||
// Validate the parameters
|
||||
// Validate the sanitized parameters
|
||||
let errors: Vec<String> = compiled_schema
|
||||
.iter_errors(params)
|
||||
.iter_errors(&sanitized)
|
||||
.map(|e| {
|
||||
let path = e.instance_path().to_string();
|
||||
if path.is_empty() {
|
||||
@@ -94,9 +253,10 @@ mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn test_validate_trigger_params_with_no_schema() {
|
||||
let trigger = Trigger {
|
||||
// ── Helper builders ──────────────────────────────────────────────
|
||||
|
||||
fn make_trigger(schema: Option<Value>) -> Trigger {
|
||||
Trigger {
|
||||
id: 1,
|
||||
r#ref: "test.trigger".to_string(),
|
||||
pack: Some(1),
|
||||
@@ -104,7 +264,7 @@ mod tests {
|
||||
label: "Test Trigger".to_string(),
|
||||
description: None,
|
||||
enabled: true,
|
||||
param_schema: None,
|
||||
param_schema: schema,
|
||||
out_schema: None,
|
||||
webhook_enabled: false,
|
||||
webhook_key: None,
|
||||
@@ -112,12 +272,43 @@ mod tests {
|
||||
is_adhoc: false,
|
||||
created: chrono::Utc::now(),
|
||||
updated: chrono::Utc::now(),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
fn make_action(schema: Option<Value>) -> Action {
|
||||
Action {
|
||||
id: 1,
|
||||
r#ref: "test.action".to_string(),
|
||||
pack: 1,
|
||||
pack_ref: "test".to_string(),
|
||||
label: "Test Action".to_string(),
|
||||
description: "Test action".to_string(),
|
||||
entrypoint: "test.sh".to_string(),
|
||||
runtime: Some(1),
|
||||
param_schema: schema,
|
||||
out_schema: None,
|
||||
is_workflow: false,
|
||||
workflow_def: None,
|
||||
is_adhoc: false,
|
||||
parameter_delivery: attune_common::models::ParameterDelivery::default(),
|
||||
parameter_format: attune_common::models::ParameterFormat::default(),
|
||||
output_format: attune_common::models::OutputFormat::default(),
|
||||
created: chrono::Utc::now(),
|
||||
updated: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
// ── No schema ────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_validate_trigger_params_with_no_schema() {
|
||||
let trigger = make_trigger(None);
|
||||
let params = json!({ "any": "value" });
|
||||
assert!(validate_trigger_params(&trigger, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
// ── Basic trigger validation (no templates) ──────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_validate_trigger_params_with_valid_params() {
|
||||
let schema = json!({
|
||||
@@ -129,24 +320,7 @@ mod tests {
|
||||
"required": ["unit", "delta"]
|
||||
});
|
||||
|
||||
let trigger = Trigger {
|
||||
id: 1,
|
||||
r#ref: "test.trigger".to_string(),
|
||||
pack: Some(1),
|
||||
pack_ref: Some("test".to_string()),
|
||||
label: "Test Trigger".to_string(),
|
||||
description: None,
|
||||
enabled: true,
|
||||
param_schema: Some(schema),
|
||||
out_schema: None,
|
||||
webhook_enabled: false,
|
||||
webhook_key: None,
|
||||
webhook_config: None,
|
||||
is_adhoc: false,
|
||||
created: chrono::Utc::now(),
|
||||
updated: chrono::Utc::now(),
|
||||
};
|
||||
|
||||
let trigger = make_trigger(Some(schema));
|
||||
let params = json!({ "unit": "seconds", "delta": 10 });
|
||||
assert!(validate_trigger_params(&trigger, ¶ms).is_ok());
|
||||
}
|
||||
@@ -162,23 +336,7 @@ mod tests {
|
||||
"required": ["unit", "delta"]
|
||||
});
|
||||
|
||||
let trigger = Trigger {
|
||||
id: 1,
|
||||
r#ref: "test.trigger".to_string(),
|
||||
pack: Some(1),
|
||||
pack_ref: Some("test".to_string()),
|
||||
label: "Test Trigger".to_string(),
|
||||
description: None,
|
||||
enabled: true,
|
||||
param_schema: Some(schema),
|
||||
out_schema: None,
|
||||
webhook_enabled: false,
|
||||
webhook_key: None,
|
||||
webhook_config: None,
|
||||
is_adhoc: false,
|
||||
created: chrono::Utc::now(),
|
||||
updated: chrono::Utc::now(),
|
||||
};
|
||||
let trigger = make_trigger(Some(schema));
|
||||
|
||||
// Missing required field 'delta'
|
||||
let params = json!({ "unit": "seconds" });
|
||||
@@ -193,6 +351,8 @@ mod tests {
|
||||
assert!(validate_trigger_params(&trigger, ¶ms).is_err());
|
||||
}
|
||||
|
||||
// ── Basic action validation (no templates) ───────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_validate_action_params_with_valid_params() {
|
||||
let schema = json!({
|
||||
@@ -203,27 +363,7 @@ mod tests {
|
||||
"required": ["message"]
|
||||
});
|
||||
|
||||
let action = Action {
|
||||
id: 1,
|
||||
r#ref: "test.action".to_string(),
|
||||
pack: 1,
|
||||
pack_ref: "test".to_string(),
|
||||
label: "Test Action".to_string(),
|
||||
description: "Test action".to_string(),
|
||||
entrypoint: "test.sh".to_string(),
|
||||
runtime: Some(1),
|
||||
param_schema: Some(schema),
|
||||
out_schema: None,
|
||||
is_workflow: false,
|
||||
workflow_def: None,
|
||||
is_adhoc: false,
|
||||
parameter_delivery: attune_common::models::ParameterDelivery::default(),
|
||||
parameter_format: attune_common::models::ParameterFormat::default(),
|
||||
output_format: attune_common::models::OutputFormat::default(),
|
||||
created: chrono::Utc::now(),
|
||||
updated: chrono::Utc::now(),
|
||||
};
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({ "message": "Hello, world!" });
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
@@ -238,28 +378,327 @@ mod tests {
|
||||
"required": ["message"]
|
||||
});
|
||||
|
||||
let action = Action {
|
||||
id: 2,
|
||||
r#ref: "test.action".to_string(),
|
||||
pack: 1,
|
||||
pack_ref: "test".to_string(),
|
||||
label: "Test Action".to_string(),
|
||||
description: "Test action".to_string(),
|
||||
entrypoint: "test.sh".to_string(),
|
||||
runtime: Some(1),
|
||||
param_schema: Some(schema),
|
||||
out_schema: None,
|
||||
is_workflow: false,
|
||||
workflow_def: None,
|
||||
is_adhoc: false,
|
||||
parameter_delivery: attune_common::models::ParameterDelivery::default(),
|
||||
parameter_format: attune_common::models::ParameterFormat::default(),
|
||||
output_format: attune_common::models::OutputFormat::default(),
|
||||
created: chrono::Utc::now(),
|
||||
updated: chrono::Utc::now(),
|
||||
};
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({});
|
||||
assert!(validate_action_params(&action, ¶ms).is_err());
|
||||
}
|
||||
|
||||
// ── Template-aware validation ────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_template_in_integer_field_passes() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"counter": { "type": "integer" }
|
||||
},
|
||||
"required": ["counter"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({ "counter": "{{ event.payload.counter }}" });
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_template_in_boolean_field_passes() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"verbose": { "type": "boolean" }
|
||||
},
|
||||
"required": ["verbose"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({ "verbose": "{{ event.payload.debug }}" });
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_template_in_number_field_passes() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"threshold": { "type": "number", "minimum": 0.0 }
|
||||
},
|
||||
"required": ["threshold"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({ "threshold": "{{ event.payload.threshold }}" });
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_template_in_enum_field_passes() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"level": { "type": "string", "enum": ["info", "warn", "error"] }
|
||||
},
|
||||
"required": ["level"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({ "level": "{{ event.payload.severity }}" });
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_template_in_array_field_passes() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"recipients": { "type": "array", "items": { "type": "string" } }
|
||||
},
|
||||
"required": ["recipients"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({ "recipients": "{{ event.payload.emails }}" });
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_template_in_object_field_passes() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"metadata": { "type": "object" }
|
||||
},
|
||||
"required": ["metadata"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({ "metadata": "{{ event.payload.meta }}" });
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_mixed_template_and_literal_values() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"message": { "type": "string" },
|
||||
"count": { "type": "integer" },
|
||||
"verbose": { "type": "boolean" }
|
||||
},
|
||||
"required": ["message", "count", "verbose"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
|
||||
// Mix of literal and template values
|
||||
let params = json!({
|
||||
"message": "Hello",
|
||||
"count": "{{ event.payload.count }}",
|
||||
"verbose": true
|
||||
});
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_literal_values_still_validated() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"message": { "type": "string" },
|
||||
"count": { "type": "integer" }
|
||||
},
|
||||
"required": ["message", "count"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
|
||||
// Template for message is fine, but literal "not_a_number" for integer is not
|
||||
let params = json!({
|
||||
"message": "{{ event.payload.msg }}",
|
||||
"count": "not_a_number"
|
||||
});
|
||||
assert!(validate_action_params(&action, ¶ms).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_required_field_still_enforced_with_templates() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"message": { "type": "string" },
|
||||
"count": { "type": "integer" }
|
||||
},
|
||||
"required": ["message", "count"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
|
||||
// Only message provided (even as template), count is missing
|
||||
let params = json!({ "message": "{{ event.payload.msg }}" });
|
||||
assert!(validate_action_params(&action, ¶ms).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_pack_config_template_passes() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"api_key": { "type": "string" },
|
||||
"timeout": { "type": "integer" }
|
||||
},
|
||||
"required": ["api_key", "timeout"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({
|
||||
"api_key": "{{ pack.config.api_key }}",
|
||||
"timeout": "{{ pack.config.default_timeout }}"
|
||||
});
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_system_template_passes() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"timestamp": { "type": "string" },
|
||||
"rule_id": { "type": "integer" }
|
||||
},
|
||||
"required": ["timestamp", "rule_id"]
|
||||
});
|
||||
|
||||
let action = make_action(Some(schema));
|
||||
let params = json!({
|
||||
"timestamp": "{{ system.timestamp }}",
|
||||
"rule_id": "{{ system.rule.id }}"
|
||||
});
|
||||
assert!(validate_action_params(&action, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_trigger_params_template_aware() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"unit": { "type": "string", "enum": ["seconds", "minutes", "hours"] },
|
||||
"delta": { "type": "integer", "minimum": 1 }
|
||||
},
|
||||
"required": ["unit", "delta"]
|
||||
});
|
||||
|
||||
let trigger = make_trigger(Some(schema));
|
||||
|
||||
// Both fields as templates
|
||||
let params = json!({
|
||||
"unit": "{{ pack.config.timer_unit }}",
|
||||
"delta": "{{ pack.config.timer_delta }}"
|
||||
});
|
||||
assert!(validate_trigger_params(&trigger, ¶ms).is_ok());
|
||||
}
|
||||
|
||||
// ── Placeholder generation ───────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_is_template_expression() {
|
||||
assert!(is_template_expression(&json!("{{ event.payload.x }}")));
|
||||
assert!(is_template_expression(&json!("{{ pack.config.key }}")));
|
||||
assert!(is_template_expression(&json!(
|
||||
"prefix {{ system.ts }} suffix"
|
||||
)));
|
||||
assert!(!is_template_expression(&json!("no braces here")));
|
||||
assert!(!is_template_expression(&json!(42)));
|
||||
assert!(!is_template_expression(&json!(true)));
|
||||
assert!(!is_template_expression(&json!("{ single braces }")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_placeholder_for_schema_types() {
|
||||
assert_eq!(
|
||||
placeholder_for_schema(&json!({"type": "integer"})),
|
||||
json!(0)
|
||||
);
|
||||
assert_eq!(
|
||||
placeholder_for_schema(&json!({"type": "number"})),
|
||||
json!(0.0)
|
||||
);
|
||||
assert_eq!(
|
||||
placeholder_for_schema(&json!({"type": "boolean"})),
|
||||
json!(true)
|
||||
);
|
||||
assert_eq!(placeholder_for_schema(&json!({"type": "array"})), json!([]));
|
||||
assert_eq!(
|
||||
placeholder_for_schema(&json!({"type": "object"})),
|
||||
json!({})
|
||||
);
|
||||
assert_eq!(
|
||||
placeholder_for_schema(&json!({"type": "string"})),
|
||||
json!("__template_placeholder__")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_placeholder_respects_enum() {
|
||||
let schema = json!({"type": "string", "enum": ["a", "b", "c"]});
|
||||
assert_eq!(placeholder_for_schema(&schema), json!("a"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_placeholder_respects_default() {
|
||||
let schema = json!({"type": "integer", "default": 42});
|
||||
assert_eq!(placeholder_for_schema(&schema), json!(42));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_placeholder_respects_minimum() {
|
||||
let schema = json!({"type": "integer", "minimum": 5});
|
||||
assert_eq!(placeholder_for_schema(&schema), json!(5));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_nested_object_template_replacement() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"outer": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"inner_count": { "type": "integer" }
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let params = json!({
|
||||
"outer": {
|
||||
"inner_count": "{{ event.payload.count }}"
|
||||
}
|
||||
});
|
||||
|
||||
let sanitized = replace_templates_with_placeholders(¶ms, &schema);
|
||||
// The inner template should be replaced with an integer placeholder
|
||||
assert!(sanitized["outer"]["inner_count"].is_number());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_array_element_template_replacement() {
|
||||
let schema = json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"tags": {
|
||||
"type": "array",
|
||||
"items": { "type": "string" }
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let params = json!({
|
||||
"tags": ["literal", "{{ event.payload.tag }}"]
|
||||
});
|
||||
|
||||
let sanitized = replace_templates_with_placeholders(¶ms, &schema);
|
||||
let tags = sanitized["tags"].as_array().unwrap();
|
||||
assert_eq!(tags[0], "literal");
|
||||
assert!(tags[1].is_string());
|
||||
assert_ne!(tags[1], "{{ event.payload.tag }}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -464,7 +464,7 @@ pub mod runtime {
|
||||
}
|
||||
|
||||
fn default_interpreter_binary() -> String {
|
||||
"/bin/sh".to_string()
|
||||
String::new()
|
||||
}
|
||||
|
||||
impl Default for InterpreterConfig {
|
||||
|
||||
@@ -144,10 +144,7 @@ impl<'a> PackComponentLoader<'a> {
|
||||
let runtime_ref = match data.get("ref").and_then(|v| v.as_str()) {
|
||||
Some(r) => r.to_string(),
|
||||
None => {
|
||||
let msg = format!(
|
||||
"Runtime YAML {} missing 'ref' field, skipping",
|
||||
filename
|
||||
);
|
||||
let msg = format!("Runtime YAML {} missing 'ref' field, skipping", filename);
|
||||
warn!("{}", msg);
|
||||
result.warnings.push(msg);
|
||||
continue;
|
||||
@@ -155,9 +152,7 @@ impl<'a> PackComponentLoader<'a> {
|
||||
};
|
||||
|
||||
// Check if runtime already exists
|
||||
if let Some(existing) =
|
||||
RuntimeRepository::find_by_ref(self.pool, &runtime_ref).await?
|
||||
{
|
||||
if let Some(existing) = RuntimeRepository::find_by_ref(self.pool, &runtime_ref).await? {
|
||||
info!(
|
||||
"Runtime '{}' already exists (ID: {}), skipping",
|
||||
runtime_ref, existing.id
|
||||
@@ -204,10 +199,7 @@ impl<'a> PackComponentLoader<'a> {
|
||||
|
||||
match RuntimeRepository::create(self.pool, input).await {
|
||||
Ok(rt) => {
|
||||
info!(
|
||||
"Created runtime '{}' (ID: {})",
|
||||
runtime_ref, rt.id
|
||||
);
|
||||
info!("Created runtime '{}' (ID: {})", runtime_ref, rt.id);
|
||||
result.runtimes_loaded += 1;
|
||||
}
|
||||
Err(e) => {
|
||||
@@ -509,15 +501,19 @@ impl<'a> PackComponentLoader<'a> {
|
||||
self.pack_ref
|
||||
);
|
||||
|
||||
// Resolve sensor runtime
|
||||
let sensor_runtime_id = self.resolve_runtime_id("builtin").await?;
|
||||
let sensor_runtime_ref = "core.builtin".to_string();
|
||||
|
||||
for (filename, content) in &yaml_files {
|
||||
let data: serde_yaml_ng::Value = serde_yaml_ng::from_str(content).map_err(|e| {
|
||||
Error::validation(format!("Failed to parse sensor YAML {}: {}", filename, e))
|
||||
})?;
|
||||
|
||||
// Resolve sensor runtime from YAML runner_type field.
|
||||
// Defaults to "native" if not specified (compiled binary, no interpreter).
|
||||
let runner_type = data
|
||||
.get("runner_type")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("native");
|
||||
let (sensor_runtime_id, sensor_runtime_ref) = self.resolve_runtime(runner_type).await?;
|
||||
|
||||
let sensor_ref = match data.get("ref").and_then(|v| v.as_str()) {
|
||||
Some(r) => r.to_string(),
|
||||
None => {
|
||||
@@ -581,7 +577,7 @@ impl<'a> PackComponentLoader<'a> {
|
||||
label,
|
||||
description,
|
||||
entrypoint,
|
||||
runtime: sensor_runtime_id.unwrap_or(0),
|
||||
runtime: sensor_runtime_id,
|
||||
runtime_ref: sensor_runtime_ref.clone(),
|
||||
trigger: trigger_id.unwrap_or(0),
|
||||
trigger_ref: trigger_ref.unwrap_or_default(),
|
||||
@@ -606,7 +602,7 @@ impl<'a> PackComponentLoader<'a> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Resolve a runtime ID from a runner type string (e.g., "shell", "python", "builtin").
|
||||
/// Resolve a runtime ID from a runner type string (e.g., "shell", "python", "native").
|
||||
///
|
||||
/// Looks up the runtime in the database by `core.{name}` ref pattern,
|
||||
/// then falls back to name-based lookup (case-insensitive).
|
||||
@@ -614,8 +610,20 @@ impl<'a> PackComponentLoader<'a> {
|
||||
/// - "shell" -> "core.shell"
|
||||
/// - "python" -> "core.python"
|
||||
/// - "node" -> "core.nodejs"
|
||||
/// - "builtin" -> "core.builtin"
|
||||
/// - "native" -> "core.native"
|
||||
async fn resolve_runtime_id(&self, runner_type: &str) -> Result<Option<Id>> {
|
||||
let (id, _ref) = self.resolve_runtime(runner_type).await?;
|
||||
if id == 0 {
|
||||
Ok(None)
|
||||
} else {
|
||||
Ok(Some(id))
|
||||
}
|
||||
}
|
||||
|
||||
/// Map a runner_type string to a (runtime_id, runtime_ref) pair.
|
||||
///
|
||||
/// Returns `(0, "unknown")` when no matching runtime is found.
|
||||
async fn resolve_runtime(&self, runner_type: &str) -> Result<(Id, String)> {
|
||||
let runner_lower = runner_type.to_lowercase();
|
||||
|
||||
// Runtime refs use the format `{pack_ref}.{name}` (e.g., "core.python").
|
||||
@@ -623,28 +631,27 @@ impl<'a> PackComponentLoader<'a> {
|
||||
"shell" | "bash" | "sh" => vec!["core.shell"],
|
||||
"python" | "python3" => vec!["core.python"],
|
||||
"node" | "nodejs" | "node.js" => vec!["core.nodejs"],
|
||||
"native" => vec!["core.native"],
|
||||
"builtin" => vec!["core.builtin"],
|
||||
"native" | "builtin" | "standalone" => vec!["core.native"],
|
||||
other => vec![other],
|
||||
};
|
||||
|
||||
for runtime_ref in &refs_to_try {
|
||||
if let Some(runtime) = RuntimeRepository::find_by_ref(self.pool, runtime_ref).await? {
|
||||
return Ok(Some(runtime.id));
|
||||
return Ok((runtime.id, runtime.r#ref));
|
||||
}
|
||||
}
|
||||
|
||||
// Fall back to name-based lookup (case-insensitive)
|
||||
use crate::repositories::runtime::RuntimeRepository as RR;
|
||||
if let Some(runtime) = RR::find_by_name(self.pool, &runner_lower).await? {
|
||||
return Ok(Some(runtime.id));
|
||||
return Ok((runtime.id, runtime.r#ref));
|
||||
}
|
||||
|
||||
warn!(
|
||||
"Could not find runtime for runner_type '{}', action will have no runtime",
|
||||
"Could not find runtime for runner_type '{}', component will have no runtime",
|
||||
runner_type
|
||||
);
|
||||
Ok(None)
|
||||
Ok((0, "unknown".to_string()))
|
||||
}
|
||||
|
||||
/// Resolve the trigger reference and ID for a sensor.
|
||||
|
||||
@@ -146,7 +146,7 @@ impl RuntimeDetector {
|
||||
|
||||
/// Verify if a runtime is available on this system
|
||||
pub async fn verify_runtime_available(runtime: &Runtime) -> bool {
|
||||
// Check if runtime is always available (e.g., shell, native, builtin)
|
||||
// Check if runtime is always available (e.g., shell, native)
|
||||
if let Some(verification) = runtime.distributions.get("verification") {
|
||||
if let Some(always_available) = verification.get("always_available") {
|
||||
if always_available.as_bool() == Some(true) {
|
||||
|
||||
@@ -264,7 +264,7 @@ mod tests {
|
||||
assert!(RefValidator::validate_runtime_ref("core.python").is_ok());
|
||||
assert!(RefValidator::validate_runtime_ref("core.shell").is_ok());
|
||||
assert!(RefValidator::validate_runtime_ref("mypack.nodejs").is_ok());
|
||||
assert!(RefValidator::validate_runtime_ref("core.builtin").is_ok());
|
||||
assert!(RefValidator::validate_runtime_ref("core.native").is_ok());
|
||||
|
||||
// Invalid formats
|
||||
assert!(RefValidator::validate_runtime_ref("core.action.webhook").is_err()); // 3-part no longer valid
|
||||
|
||||
@@ -17,6 +17,7 @@ use attune_common::{
|
||||
},
|
||||
repositories::{
|
||||
event::{CreateEnforcementInput, EnforcementRepository, EventRepository},
|
||||
pack::PackRepository,
|
||||
rule::RuleRepository,
|
||||
Create, FindById, List,
|
||||
},
|
||||
@@ -191,7 +192,7 @@ impl EventProcessor {
|
||||
.unwrap_or_else(|| serde_json::Map::new());
|
||||
|
||||
// Resolve action parameters using the template resolver
|
||||
let resolved_params = Self::resolve_action_params(rule, event, &payload)?;
|
||||
let resolved_params = Self::resolve_action_params(pool, rule, event, &payload).await?;
|
||||
|
||||
let create_input = CreateEnforcementInput {
|
||||
rule: Some(rule.id),
|
||||
@@ -354,7 +355,8 @@ impl EventProcessor {
|
||||
/// Replaces `{{ event.payload.* }}`, `{{ event.id }}`, `{{ event.trigger }}`,
|
||||
/// `{{ event.created }}`, `{{ pack.config.* }}`, and `{{ system.* }}` references
|
||||
/// in the rule's `action_params` with values from the event and context.
|
||||
fn resolve_action_params(
|
||||
async fn resolve_action_params(
|
||||
pool: &PgPool,
|
||||
rule: &Rule,
|
||||
event: &Event,
|
||||
event_payload: &serde_json::Value,
|
||||
@@ -366,11 +368,26 @@ impl EventProcessor {
|
||||
return Ok(serde_json::Map::new());
|
||||
}
|
||||
|
||||
// Load pack config from database for pack.config.* resolution
|
||||
let pack_config = match PackRepository::find_by_id(pool, rule.pack).await {
|
||||
Ok(Some(pack)) => pack.config,
|
||||
Ok(None) => {
|
||||
warn!(
|
||||
"Pack {} not found for rule {} — pack.config.* templates will resolve to null",
|
||||
rule.pack, rule.r#ref
|
||||
);
|
||||
serde_json::json!({})
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Failed to load pack {} for rule {}: {} — pack.config.* templates will resolve to null", rule.pack, rule.r#ref, e);
|
||||
serde_json::json!({})
|
||||
}
|
||||
};
|
||||
|
||||
// Build template context from the event
|
||||
let context = TemplateContext::new(
|
||||
event_payload.clone(),
|
||||
// TODO: Load pack config from database for pack.config.* resolution
|
||||
serde_json::json!({}),
|
||||
pack_config,
|
||||
serde_json::json!({
|
||||
"timestamp": chrono::Utc::now().to_rfc3339(),
|
||||
"rule": {
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
|
||||
use anyhow::{anyhow, Result};
|
||||
use attune_common::models::{Id, Sensor, Trigger};
|
||||
use attune_common::repositories::{FindById, List};
|
||||
use attune_common::repositories::{FindById, List, RuntimeRepository};
|
||||
|
||||
use sqlx::{PgPool, Row};
|
||||
use std::collections::HashMap;
|
||||
@@ -38,6 +38,7 @@ struct SensorManagerInner {
|
||||
sensors: Arc<RwLock<HashMap<Id, SensorInstance>>>,
|
||||
running: Arc<RwLock<bool>>,
|
||||
packs_base_dir: String,
|
||||
runtime_envs_dir: String,
|
||||
api_client: ApiClient,
|
||||
api_url: String,
|
||||
mq_url: String,
|
||||
@@ -58,6 +59,10 @@ impl SensorManager {
|
||||
let mq_url = std::env::var("ATTUNE_MQ_URL")
|
||||
.unwrap_or_else(|_| "amqp://guest:guest@localhost:5672".to_string());
|
||||
|
||||
let runtime_envs_dir = std::env::var("ATTUNE_RUNTIME_ENVS_DIR")
|
||||
.or_else(|_| std::env::var("ATTUNE__RUNTIME_ENVS_DIR"))
|
||||
.unwrap_or_else(|_| "/opt/attune/runtime_envs".to_string());
|
||||
|
||||
// Create API client for token provisioning (no admin token - uses internal endpoint)
|
||||
let api_client = ApiClient::new(api_url.clone(), None);
|
||||
|
||||
@@ -67,6 +72,7 @@ impl SensorManager {
|
||||
sensors: Arc::new(RwLock::new(HashMap::new())),
|
||||
running: Arc::new(RwLock::new(false)),
|
||||
packs_base_dir,
|
||||
runtime_envs_dir,
|
||||
api_client,
|
||||
api_url,
|
||||
mq_url,
|
||||
@@ -212,9 +218,45 @@ impl SensorManager {
|
||||
self.inner.packs_base_dir, pack_ref, sensor.entrypoint
|
||||
);
|
||||
|
||||
// Load the runtime to determine how to execute the sensor
|
||||
let runtime = RuntimeRepository::find_by_id(&self.inner.db, sensor.runtime)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
anyhow!(
|
||||
"Runtime {} not found for sensor {}",
|
||||
sensor.runtime,
|
||||
sensor.r#ref
|
||||
)
|
||||
})?;
|
||||
|
||||
let exec_config = runtime.parsed_execution_config();
|
||||
let rt_name = runtime.name.to_lowercase();
|
||||
|
||||
// Resolve the interpreter: check for a virtualenv/node_modules first,
|
||||
// then fall back to the system interpreter.
|
||||
let pack_dir = std::path::PathBuf::from(&self.inner.packs_base_dir).join(pack_ref);
|
||||
let env_dir = std::path::PathBuf::from(&self.inner.runtime_envs_dir)
|
||||
.join(pack_ref)
|
||||
.join(&rt_name);
|
||||
let env_dir_opt = if env_dir.exists() {
|
||||
Some(env_dir.as_path())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// Determine whether we need an interpreter or can execute directly.
|
||||
// Determine native vs interpreted purely from the runtime's execution_config.
|
||||
// A native runtime (e.g., core.native) has no interpreter configured —
|
||||
// its binary field is empty. Interpreted runtimes (Python, Node, etc.)
|
||||
// declare their interpreter binary explicitly in execution_config.
|
||||
let interpreter_binary = &exec_config.interpreter.binary;
|
||||
let is_native = interpreter_binary.is_empty()
|
||||
|| interpreter_binary == "native"
|
||||
|| interpreter_binary == "none";
|
||||
|
||||
info!(
|
||||
"TRACE: Before fetching trigger instances for sensor {}",
|
||||
sensor.r#ref
|
||||
"Sensor {} runtime={} interpreter={} native={}",
|
||||
sensor.r#ref, rt_name, interpreter_binary, is_native
|
||||
);
|
||||
info!("Starting standalone sensor process: {}", sensor_script);
|
||||
|
||||
@@ -245,9 +287,30 @@ impl SensorManager {
|
||||
.map_err(|e| anyhow!("Failed to serialize trigger instances: {}", e))?;
|
||||
info!("Trigger instances JSON: {}", trigger_instances_json);
|
||||
|
||||
// Build the command: use the interpreter for non-native runtimes,
|
||||
// execute the script directly for native binaries.
|
||||
let mut cmd = if is_native {
|
||||
Command::new(&sensor_script)
|
||||
} else {
|
||||
let resolved_interpreter =
|
||||
exec_config.resolve_interpreter_with_env(&pack_dir, env_dir_opt);
|
||||
info!(
|
||||
"Using interpreter {} for sensor {}",
|
||||
resolved_interpreter.display(),
|
||||
sensor.r#ref
|
||||
);
|
||||
let mut c = Command::new(resolved_interpreter);
|
||||
// Pass any extra interpreter args (e.g., -u for unbuffered Python)
|
||||
for arg in &exec_config.interpreter.args {
|
||||
c.arg(arg);
|
||||
}
|
||||
c.arg(&sensor_script);
|
||||
c
|
||||
};
|
||||
|
||||
// Start the standalone sensor with token and configuration
|
||||
// Pass sensor ref (e.g., "core.interval_timer_sensor") for proper identification
|
||||
let mut child = Command::new(&sensor_script)
|
||||
let mut child = cmd
|
||||
.env("ATTUNE_API_URL", &self.inner.api_url)
|
||||
.env("ATTUNE_API_TOKEN", &token_response.token)
|
||||
.env("ATTUNE_SENSOR_ID", &sensor.id.to_string())
|
||||
|
||||
@@ -434,12 +434,18 @@ async fn process_runtime_for_pack(
|
||||
///
|
||||
/// Returns `None` if the variable is not set (meaning all runtimes are accepted).
|
||||
pub fn runtime_filter_from_env() -> Option<Vec<String>> {
|
||||
std::env::var("ATTUNE_WORKER_RUNTIMES").ok().map(|val| {
|
||||
val.split(',')
|
||||
.map(|s| s.trim().to_lowercase())
|
||||
.filter(|s| !s.is_empty())
|
||||
.collect()
|
||||
})
|
||||
std::env::var("ATTUNE_WORKER_RUNTIMES")
|
||||
.ok()
|
||||
.map(|val| parse_runtime_filter(&val))
|
||||
}
|
||||
|
||||
/// Parse a comma-separated runtime filter string into a list of lowercase runtime names.
|
||||
/// Empty entries are filtered out.
|
||||
fn parse_runtime_filter(val: &str) -> Vec<String> {
|
||||
val.split(',')
|
||||
.map(|s| s.trim().to_lowercase())
|
||||
.filter(|s| !s.is_empty())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -447,26 +453,21 @@ mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_runtime_filter_from_env_not_set() {
|
||||
// When ATTUNE_WORKER_RUNTIMES is not set, filter should be None
|
||||
std::env::remove_var("ATTUNE_WORKER_RUNTIMES");
|
||||
assert!(runtime_filter_from_env().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_runtime_filter_from_env_set() {
|
||||
std::env::set_var("ATTUNE_WORKER_RUNTIMES", "shell,Python, Node");
|
||||
let filter = runtime_filter_from_env().unwrap();
|
||||
fn test_parse_runtime_filter_values() {
|
||||
let filter = parse_runtime_filter("shell,Python, Node");
|
||||
assert_eq!(filter, vec!["shell", "python", "node"]);
|
||||
std::env::remove_var("ATTUNE_WORKER_RUNTIMES");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_runtime_filter_from_env_empty() {
|
||||
std::env::set_var("ATTUNE_WORKER_RUNTIMES", "");
|
||||
let filter = runtime_filter_from_env().unwrap();
|
||||
fn test_parse_runtime_filter_empty() {
|
||||
let filter = parse_runtime_filter("");
|
||||
assert!(filter.is_empty());
|
||||
std::env::remove_var("ATTUNE_WORKER_RUNTIMES");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parse_runtime_filter_whitespace() {
|
||||
let filter = parse_runtime_filter(" shell , , python ");
|
||||
assert_eq!(filter, vec!["shell", "python"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -55,12 +55,20 @@ pub async fn execute_streaming(
|
||||
let stdin_write_error = if let Some(mut stdin) = child.stdin.take() {
|
||||
let mut error = None;
|
||||
|
||||
// Write parameters first if using stdin delivery
|
||||
// Write parameters first if using stdin delivery.
|
||||
// Skip empty/trivial content ("{}","","[]") to avoid polluting stdin
|
||||
// before secrets — scripts that read secrets via readline() expect
|
||||
// the secrets JSON as the first line.
|
||||
let has_real_params = parameters_stdin
|
||||
.map(|s| !matches!(s.trim(), "" | "{}" | "[]"))
|
||||
.unwrap_or(false);
|
||||
if let Some(params_data) = parameters_stdin {
|
||||
if let Err(e) = stdin.write_all(params_data.as_bytes()).await {
|
||||
error = Some(format!("Failed to write parameters to stdin: {}", e));
|
||||
} else if let Err(e) = stdin.write_all(b"\n---ATTUNE_PARAMS_END---\n").await {
|
||||
error = Some(format!("Failed to write parameter delimiter: {}", e));
|
||||
if has_real_params {
|
||||
if let Err(e) = stdin.write_all(params_data.as_bytes()).await {
|
||||
error = Some(format!("Failed to write parameters to stdin: {}", e));
|
||||
} else if let Err(e) = stdin.write_all(b"\n---ATTUNE_PARAMS_END---\n").await {
|
||||
error = Some(format!("Failed to write parameter delimiter: {}", e));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ use super::{
|
||||
RuntimeResult,
|
||||
};
|
||||
use async_trait::async_trait;
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::process::Stdio;
|
||||
use std::time::Instant;
|
||||
@@ -16,6 +17,15 @@ use tokio::process::Command;
|
||||
use tokio::time::timeout;
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
/// Escape a string for embedding inside a bash single-quoted string.
|
||||
///
|
||||
/// In single-quoted strings the only problematic character is `'` itself.
|
||||
/// We close the current single-quote, insert an escaped single-quote, and
|
||||
/// reopen: `'foo'\''bar'` → `foo'bar`.
|
||||
fn bash_single_quote_escape(s: &str) -> String {
|
||||
s.replace('\'', "'\\''")
|
||||
}
|
||||
|
||||
/// Shell runtime for executing shell scripts and commands
|
||||
pub struct ShellRuntime {
|
||||
/// Shell interpreter path (bash, sh, zsh, etc.)
|
||||
@@ -75,12 +85,20 @@ impl ShellRuntime {
|
||||
let stdin_write_error = if let Some(mut stdin) = child.stdin.take() {
|
||||
let mut error = None;
|
||||
|
||||
// Write parameters first if using stdin delivery
|
||||
// Write parameters first if using stdin delivery.
|
||||
// Skip empty/trivial content ("{}","","[]") to avoid polluting stdin
|
||||
// before secrets — scripts that read secrets via readline() expect
|
||||
// the secrets JSON as the first line.
|
||||
let has_real_params = parameters_stdin
|
||||
.map(|s| !matches!(s.trim(), "" | "{}" | "[]"))
|
||||
.unwrap_or(false);
|
||||
if let Some(params_data) = parameters_stdin {
|
||||
if let Err(e) = stdin.write_all(params_data.as_bytes()).await {
|
||||
error = Some(format!("Failed to write parameters to stdin: {}", e));
|
||||
} else if let Err(e) = stdin.write_all(b"\n---ATTUNE_PARAMS_END---\n").await {
|
||||
error = Some(format!("Failed to write parameter delimiter: {}", e));
|
||||
if has_real_params {
|
||||
if let Err(e) = stdin.write_all(params_data.as_bytes()).await {
|
||||
error = Some(format!("Failed to write parameters to stdin: {}", e));
|
||||
} else if let Err(e) = stdin.write_all(b"\n---ATTUNE_PARAMS_END---\n").await {
|
||||
error = Some(format!("Failed to write parameter delimiter: {}", e));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -300,7 +318,12 @@ impl ShellRuntime {
|
||||
})
|
||||
}
|
||||
|
||||
/// Generate shell wrapper script that injects parameters as environment variables
|
||||
/// Generate shell wrapper script that injects parameters and secrets directly.
|
||||
///
|
||||
/// Secrets are embedded as bash associative-array entries at generation time
|
||||
/// so the wrapper has **zero external runtime dependencies** (no Python, jq,
|
||||
/// etc.). The generated script is written to a temp file by the caller so
|
||||
/// that secrets never appear in `/proc/<pid>/cmdline`.
|
||||
fn generate_wrapper_script(&self, context: &ExecutionContext) -> RuntimeResult<String> {
|
||||
let mut script = String::new();
|
||||
|
||||
@@ -308,25 +331,19 @@ impl ShellRuntime {
|
||||
script.push_str("#!/bin/bash\n");
|
||||
script.push_str("set -e\n\n"); // Exit on error
|
||||
|
||||
// Read secrets from stdin and store in associative array
|
||||
script.push_str("# Read secrets from stdin (passed securely, not via environment)\n");
|
||||
// Populate secrets associative array directly from Rust — no stdin
|
||||
// reading, no JSON parsing, no external interpreters.
|
||||
script.push_str("# Secrets (injected at generation time, not via environment)\n");
|
||||
script.push_str("declare -A ATTUNE_SECRETS\n");
|
||||
script.push_str("read -r ATTUNE_SECRETS_JSON\n");
|
||||
script.push_str("if [ -n \"$ATTUNE_SECRETS_JSON\" ]; then\n");
|
||||
script.push_str(" # Parse JSON secrets using Python (always available)\n");
|
||||
script.push_str(" eval \"$(echo \"$ATTUNE_SECRETS_JSON\" | python3 -c \"\n");
|
||||
script.push_str("import sys, json\n");
|
||||
script.push_str("try:\n");
|
||||
script.push_str(" secrets = json.load(sys.stdin)\n");
|
||||
script.push_str(" for key, value in secrets.items():\n");
|
||||
script.push_str(" # Escape single quotes in value\n");
|
||||
script.push_str(
|
||||
" safe_value = value.replace(\\\"'\\\", \\\"'\\\\\\\\\\\\\\\\'\\\") \n",
|
||||
);
|
||||
script.push_str(" print(f\\\"ATTUNE_SECRETS['{key}']='{safe_value}'\\\")\n");
|
||||
script.push_str("except: pass\n");
|
||||
script.push_str("\")\"\n");
|
||||
script.push_str("fi\n\n");
|
||||
for (key, value) in &context.secrets {
|
||||
let escaped_key = bash_single_quote_escape(key);
|
||||
let escaped_val = bash_single_quote_escape(value);
|
||||
script.push_str(&format!(
|
||||
"ATTUNE_SECRETS['{}']='{}'\n",
|
||||
escaped_key, escaped_val
|
||||
));
|
||||
}
|
||||
script.push('\n');
|
||||
|
||||
// Helper function to get secrets
|
||||
script.push_str("# Helper function to access secrets\n");
|
||||
@@ -344,16 +361,17 @@ impl ShellRuntime {
|
||||
serde_json::Value::Bool(b) => b.to_string(),
|
||||
_ => serde_json::to_string(value)?,
|
||||
};
|
||||
let escaped = bash_single_quote_escape(&value_str);
|
||||
// Export with PARAM_ prefix for consistency
|
||||
script.push_str(&format!(
|
||||
"export PARAM_{}='{}'\n",
|
||||
key.to_uppercase(),
|
||||
value_str
|
||||
escaped
|
||||
));
|
||||
// Also export without prefix for easier shell script writing
|
||||
script.push_str(&format!("export {}='{}'\n", key, value_str));
|
||||
script.push_str(&format!("export {}='{}'\n", key, escaped));
|
||||
}
|
||||
script.push_str("\n");
|
||||
script.push('\n');
|
||||
|
||||
// Add the action code
|
||||
script.push_str("# Action code\n");
|
||||
@@ -364,44 +382,6 @@ impl ShellRuntime {
|
||||
Ok(script)
|
||||
}
|
||||
|
||||
/// Execute shell script directly
|
||||
async fn execute_shell_code(
|
||||
&self,
|
||||
code: String,
|
||||
secrets: &std::collections::HashMap<String, String>,
|
||||
env: &std::collections::HashMap<String, String>,
|
||||
parameters_stdin: Option<&str>,
|
||||
timeout_secs: Option<u64>,
|
||||
max_stdout_bytes: usize,
|
||||
max_stderr_bytes: usize,
|
||||
output_format: OutputFormat,
|
||||
) -> RuntimeResult<ExecutionResult> {
|
||||
debug!(
|
||||
"Executing shell script with {} secrets (passed via stdin)",
|
||||
secrets.len()
|
||||
);
|
||||
|
||||
// Build command
|
||||
let mut cmd = Command::new(&self.shell_path);
|
||||
cmd.arg("-c").arg(&code);
|
||||
|
||||
// Add environment variables
|
||||
for (key, value) in env {
|
||||
cmd.env(key, value);
|
||||
}
|
||||
|
||||
self.execute_with_streaming(
|
||||
cmd,
|
||||
secrets,
|
||||
parameters_stdin,
|
||||
timeout_secs,
|
||||
max_stdout_bytes,
|
||||
max_stderr_bytes,
|
||||
output_format,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Execute shell script from file
|
||||
async fn execute_shell_file(
|
||||
&self,
|
||||
@@ -520,19 +500,42 @@ impl Runtime for ShellRuntime {
|
||||
.await;
|
||||
}
|
||||
|
||||
// Otherwise, generate wrapper script and execute
|
||||
// Otherwise, generate wrapper script and execute.
|
||||
// Secrets and parameters are embedded directly in the wrapper script
|
||||
// by generate_wrapper_script(), so we write it to a temp file (to keep
|
||||
// secrets out of /proc/cmdline) and pass no secrets/params via stdin.
|
||||
let script = self.generate_wrapper_script(&context)?;
|
||||
self.execute_shell_code(
|
||||
script,
|
||||
&context.secrets,
|
||||
&env,
|
||||
parameters_stdin,
|
||||
context.timeout,
|
||||
context.max_stdout_bytes,
|
||||
context.max_stderr_bytes,
|
||||
context.output_format,
|
||||
)
|
||||
.await
|
||||
|
||||
// Write wrapper to a temp file so secrets are not exposed in the
|
||||
// process command line (which would happen with `bash -c "..."`).
|
||||
let wrapper_dir = self.work_dir.join("wrappers");
|
||||
tokio::fs::create_dir_all(&wrapper_dir).await.map_err(|e| {
|
||||
RuntimeError::ExecutionFailed(format!("Failed to create wrapper directory: {}", e))
|
||||
})?;
|
||||
let wrapper_path = wrapper_dir.join(format!("wrapper_{}.sh", context.execution_id));
|
||||
tokio::fs::write(&wrapper_path, &script)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
RuntimeError::ExecutionFailed(format!("Failed to write wrapper script: {}", e))
|
||||
})?;
|
||||
|
||||
let result = self
|
||||
.execute_shell_file(
|
||||
wrapper_path.clone(),
|
||||
&HashMap::new(), // secrets are in the script, not stdin
|
||||
&env,
|
||||
None,
|
||||
context.timeout,
|
||||
context.max_stdout_bytes,
|
||||
context.max_stderr_bytes,
|
||||
context.output_format,
|
||||
)
|
||||
.await;
|
||||
|
||||
// Clean up wrapper file (best-effort)
|
||||
let _ = tokio::fs::remove_file(&wrapper_path).await;
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
async fn setup(&self) -> RuntimeResult<()> {
|
||||
@@ -716,7 +719,6 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "Pre-existing failure - secrets not being passed correctly"]
|
||||
async fn test_shell_runtime_with_secrets() {
|
||||
let runtime = ShellRuntime::new();
|
||||
|
||||
|
||||
@@ -157,8 +157,8 @@ impl WorkerService {
|
||||
// Load runtimes from the database and create ProcessRuntime instances.
|
||||
// Each runtime row's `execution_config` JSONB drives how the ProcessRuntime
|
||||
// invokes interpreters, manages environments, and installs dependencies.
|
||||
// We skip runtimes with empty execution_config (e.g., the built-in sensor
|
||||
// runtime) since they have no interpreter and cannot execute as a process.
|
||||
// We skip runtimes with empty execution_config (e.g., core.native) since
|
||||
// they execute binaries directly and don't need a ProcessRuntime wrapper.
|
||||
match RuntimeRepository::list(&pool).await {
|
||||
Ok(db_runtimes) => {
|
||||
let executable_runtimes: Vec<_> = db_runtimes
|
||||
|
||||
Reference in New Issue
Block a user