enact_core/policy/

input_processor.rs

//! Input Processor - Pre-execution content validation
//!
//! Input processors run BEFORE execution to validate and transform user input.
//! They detect potential issues like:
//! - PII in prompts (may need masking or blocking)
//! - Prompt injection attempts
//! - Prohibited content
//!
//! ## Architecture
//!
//! ```text
//! User Input
//!     │
//!     ▼
//! ┌──────────────────────────────────────────┐
//! │        INPUT PROCESSOR PIPELINE          │
//! │ ┌──────────┐ ┌──────────┐ ┌───────────┐ │
//! │ │ Content  │→│ Prompt   │→│   PII     │ │
//! │ │ Filter   │ │ Injection│ │ Detection │ │
//! │ └──────────┘ └──────────┘ └───────────┘ │
//! │      Pass/Block/Modify                   │
//! └──────────────────────────────────────────┘
//!     │
//!     ▼ (if Pass)
//! ExecutionKernel
//! ```
//!
//! @see docs/TECHNICAL/17-GUARDRAILS-PROTECTION.md
//! @see docs/TECHNICAL/25-STREAM-PROCESSORS.md

use super::PolicyContext;
use async_trait::async_trait;
use std::sync::Arc;

/// Result of input processing
#[derive(Debug, Clone)]
pub enum InputProcessorResult {
    /// Input is safe, proceed with execution
    Pass,

    /// Input is blocked - do not execute
    Block {
        reason: String,
        /// The processor that blocked
        processor: String,
    },

    /// Input was modified (sanitized) - proceed with modified input
    Modify {
        original: String,
        modified: String,
        /// What was changed
        changes: Vec<String>,
    },
}

impl InputProcessorResult {
    /// Returns true if execution should proceed
    pub fn should_proceed(&self) -> bool {
        matches!(self, Self::Pass | Self::Modify { .. })
    }

    /// Returns true if input was blocked
    pub fn is_blocked(&self) -> bool {
        matches!(self, Self::Block { .. })
    }

    /// Get the input to use for execution (original or modified)
    pub fn effective_input<'a>(&'a self, original: &'a str) -> &'a str {
        match self {
            Self::Modify { modified, .. } => modified,
            _ => original,
        }
    }
}

/// Input Processor trait
///
/// Processors run BEFORE execution to validate/transform user input.
/// They MUST NOT have side effects beyond logging.
#[async_trait]
pub trait InputProcessor: Send + Sync {
    /// Processor name for logging/metrics
    fn name(&self) -> &str;

    /// Priority for ordering (lower = runs first)
    fn priority(&self) -> u32 {
        100 // Default priority
    }

    /// Process input before execution
    ///
    /// # Arguments
    /// * `input` - The user input to validate
    /// * `ctx` - Policy context (tenant, user, metadata)
    ///
    /// # Returns
    /// - `Pass` - Input is safe
    /// - `Block` - Input is blocked (don't execute)
    /// - `Modify` - Input was sanitized (use modified version)
    async fn process(
        &self,
        input: &str,
        ctx: &PolicyContext,
    ) -> anyhow::Result<InputProcessorResult>;
}

/// Input Processor Pipeline - chains multiple processors
pub struct InputProcessorPipeline {
    processors: Vec<Arc<dyn InputProcessor>>,
}

impl InputProcessorPipeline {
    /// Create a new empty pipeline
    pub fn new() -> Self {
        Self { processors: vec![] }
    }

    /// Add a processor to the pipeline
    #[allow(clippy::should_implement_trait)]
    pub fn add(mut self, processor: Arc<dyn InputProcessor>) -> Self {
        self.processors.push(processor);
        // Sort by priority (lower = first)
        self.processors.sort_by_key(|p| p.priority());
        self
    }

    /// Run all processors in sequence
    ///
    /// Processing stops on first Block.
    /// Modify results are chained (each processor sees the modified input).
    pub async fn process(
        &self,
        input: &str,
        ctx: &PolicyContext,
    ) -> anyhow::Result<InputProcessorResult> {
        let mut current_input = input.to_string();
        let mut all_changes: Vec<String> = vec![];
        let mut was_modified = false;

        for processor in &self.processors {
            let result = processor.process(&current_input, ctx).await?;

            match result {
                InputProcessorResult::Pass => {
                    // Continue to next processor
                    continue;
                }
                InputProcessorResult::Block { .. } => {
                    // Stop processing, return block
                    return Ok(result);
                }
                InputProcessorResult::Modify {
                    modified, changes, ..
                } => {
                    // Update input and continue
                    was_modified = true;
                    all_changes.extend(changes);
                    current_input = modified;
                }
            }
        }

        // All processors passed
        if was_modified {
            Ok(InputProcessorResult::Modify {
                original: input.to_string(),
                modified: current_input,
                changes: all_changes,
            })
        } else {
            Ok(InputProcessorResult::Pass)
        }
    }

    /// Check if the pipeline is empty
    pub fn is_empty(&self) -> bool {
        self.processors.is_empty()
    }

    /// Get the number of processors
    pub fn len(&self) -> usize {
        self.processors.len()
    }
}

impl Default for InputProcessorPipeline {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::policy::PolicyAction;
    use std::collections::HashMap;

    // Mock processor for testing
    struct MockPassProcessor;

    #[async_trait]
    impl InputProcessor for MockPassProcessor {
        fn name(&self) -> &str {
            "mock-pass"
        }

        async fn process(
            &self,
            _input: &str,
            _ctx: &PolicyContext,
        ) -> anyhow::Result<InputProcessorResult> {
            Ok(InputProcessorResult::Pass)
        }
    }

    struct MockBlockProcessor {
        reason: String,
    }

    #[async_trait]
    impl InputProcessor for MockBlockProcessor {
        fn name(&self) -> &str {
            "mock-block"
        }

        async fn process(
            &self,
            _input: &str,
            _ctx: &PolicyContext,
        ) -> anyhow::Result<InputProcessorResult> {
            Ok(InputProcessorResult::Block {
                reason: self.reason.clone(),
                processor: self.name().to_string(),
            })
        }
    }

    struct MockModifyProcessor {
        suffix: String,
    }

    #[async_trait]
    impl InputProcessor for MockModifyProcessor {
        fn name(&self) -> &str {
            "mock-modify"
        }

        async fn process(
            &self,
            input: &str,
            _ctx: &PolicyContext,
        ) -> anyhow::Result<InputProcessorResult> {
            Ok(InputProcessorResult::Modify {
                original: input.to_string(),
                modified: format!("{}{}", input, self.suffix),
                changes: vec![format!("Added suffix: {}", self.suffix)],
            })
        }
    }

    fn test_context() -> PolicyContext {
        PolicyContext {
            tenant_id: Some("test-tenant".to_string()),
            user_id: Some("test-user".to_string()),
            action: PolicyAction::StartExecution { graph_id: None },
            metadata: HashMap::new(),
        }
    }

    #[test]
    fn test_input_processor_result_should_proceed() {
        assert!(InputProcessorResult::Pass.should_proceed());
        assert!(InputProcessorResult::Modify {
            original: "a".to_string(),
            modified: "b".to_string(),
            changes: vec![],
        }
        .should_proceed());
        assert!(!InputProcessorResult::Block {
            reason: "test".to_string(),
            processor: "test".to_string(),
        }
        .should_proceed());
    }

    #[test]
    fn test_input_processor_result_is_blocked() {
        assert!(!InputProcessorResult::Pass.is_blocked());
        assert!(InputProcessorResult::Block {
            reason: "test".to_string(),
            processor: "test".to_string(),
        }
        .is_blocked());
    }

    #[test]
    fn test_input_processor_result_effective_input() {
        let original = "hello";

        // Pass returns original
        assert_eq!(
            InputProcessorResult::Pass.effective_input(original),
            "hello"
        );

        // Block returns original
        let block = InputProcessorResult::Block {
            reason: "blocked".to_string(),
            processor: "test".to_string(),
        };
        assert_eq!(block.effective_input(original), "hello");

        // Modify returns modified
        let modify = InputProcessorResult::Modify {
            original: "hello".to_string(),
            modified: "hello world".to_string(),
            changes: vec![],
        };
        assert_eq!(modify.effective_input(original), "hello world");
    }

    #[tokio::test]
    async fn test_pipeline_empty() {
        let pipeline = InputProcessorPipeline::new();
        assert!(pipeline.is_empty());
        assert_eq!(pipeline.len(), 0);

        let ctx = test_context();
        let result = pipeline.process("test input", &ctx).await.unwrap();
        assert!(matches!(result, InputProcessorResult::Pass));
    }

    #[tokio::test]
    async fn test_pipeline_pass_through() {
        let pipeline = InputProcessorPipeline::new().add(Arc::new(MockPassProcessor));

        let ctx = test_context();
        let result = pipeline.process("test input", &ctx).await.unwrap();
        assert!(matches!(result, InputProcessorResult::Pass));
    }

    #[tokio::test]
    async fn test_pipeline_block() {
        let pipeline = InputProcessorPipeline::new()
            .add(Arc::new(MockPassProcessor))
            .add(Arc::new(MockBlockProcessor {
                reason: "forbidden".to_string(),
            }));

        let ctx = test_context();
        let result = pipeline.process("test input", &ctx).await.unwrap();

        assert!(result.is_blocked());
        if let InputProcessorResult::Block { reason, processor } = result {
            assert_eq!(reason, "forbidden");
            assert_eq!(processor, "mock-block");
        }
    }

    #[tokio::test]
    async fn test_pipeline_modify() {
        let pipeline = InputProcessorPipeline::new().add(Arc::new(MockModifyProcessor {
            suffix: " [sanitized]".to_string(),
        }));

        let ctx = test_context();
        let result = pipeline.process("test input", &ctx).await.unwrap();

        if let InputProcessorResult::Modify {
            original,
            modified,
            changes,
        } = result
        {
            assert_eq!(original, "test input");
            assert_eq!(modified, "test input [sanitized]");
            assert_eq!(changes.len(), 1);
        } else {
            panic!("Expected Modify result");
        }
    }

    #[tokio::test]
    async fn test_pipeline_chained_modify() {
        let pipeline = InputProcessorPipeline::new()
            .add(Arc::new(MockModifyProcessor {
                suffix: " [a]".to_string(),
            }))
            .add(Arc::new(MockModifyProcessor {
                suffix: " [b]".to_string(),
            }));

        let ctx = test_context();
        let result = pipeline.process("input", &ctx).await.unwrap();

        if let InputProcessorResult::Modify {
            modified, changes, ..
        } = result
        {
            // Both modifications applied in sequence
            assert_eq!(modified, "input [a] [b]");
            assert_eq!(changes.len(), 2);
        } else {
            panic!("Expected Modify result");
        }
    }
}