camel_component_validator/
component.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::sync::Arc;
4use std::task::{Context, Poll};
5
6use tower::Service;
7use tracing::debug;
8
9use crate::compiled::CompiledValidator;
10use crate::config::ValidatorConfig;
11use crate::xsd_bridge::{XsdBridge, XsdBridgeBackend};
12use camel_component_api::ComponentContext;
13use camel_component_api::{
14    BoxProcessor, CamelError, Component, ComponentMetadata, Consumer, Endpoint, Exchange,
15    ProducerContext, RuntimeObservability,
16};
17
18use crate::metadata::ValidatorMetadataDescriptor;
19
20pub struct ValidatorComponent {
21    xsd_bridge: Arc<dyn XsdBridge>,
22    xsd_backend: Option<Arc<XsdBridgeBackend>>,
23}
24
25impl ValidatorComponent {
26    pub fn new() -> Self {
27        let xsd_backend = Arc::new(XsdBridgeBackend::new());
28        Self {
29            xsd_bridge: Arc::clone(&xsd_backend) as Arc<dyn XsdBridge>,
30            xsd_backend: Some(xsd_backend),
31        }
32    }
33
34    pub fn xsd_bridge_backend(&self) -> Option<Arc<XsdBridgeBackend>> {
35        self.xsd_backend.as_ref().map(Arc::clone)
36    }
37
38    #[cfg(test)]
39    fn with_xsd_bridge(xsd_bridge: Arc<dyn XsdBridge>) -> Self {
40        Self {
41            xsd_bridge,
42            xsd_backend: None,
43        }
44    }
45}
46
47impl Default for ValidatorComponent {
48    fn default() -> Self {
49        Self::new()
50    }
51}
52
53impl Component for ValidatorComponent {
54    fn scheme(&self) -> &str {
55        "validator"
56    }
57
58    fn metadata(&self) -> ComponentMetadata {
59        ValidatorMetadataDescriptor::metadata()
60    }
61
62    fn create_endpoint(
63        &self,
64        uri: &str,
65        _ctx: &dyn ComponentContext,
66    ) -> Result<Box<dyn Endpoint>, CamelError> {
67        let config = ValidatorConfig::from_uri(uri)?;
68        let xsd_bridge = Arc::clone(&self.xsd_bridge);
69        let compiled = CompiledValidator::compile(&config, xsd_bridge)?;
70        Ok(Box::new(ValidatorEndpoint {
71            uri: uri.to_string(),
72            config,
73            compiled: Arc::new(compiled),
74            xsd_backend: self.xsd_backend.as_ref().map(Arc::clone),
75        }))
76    }
77}
78
79/// Validator endpoint that checks message bodies or headers against a schema.
80///
81/// Supports XML (XSD), JSON Schema, and YAML schema validation.
82/// RelaxNG and Schematron are accepted in URI parsing but rejected at endpoint
83/// creation with a clear error message.
84struct ValidatorEndpoint {
85    uri: String,
86    config: ValidatorConfig,
87    compiled: Arc<CompiledValidator>,
88    xsd_backend: Option<Arc<XsdBridgeBackend>>,
89}
90
91impl ValidatorEndpoint {
92    /// Returns a human-readable description of the configured schema.
93    #[allow(dead_code)]
94    pub fn schema_info(&self) -> String {
95        format!(
96            "{:?} schema: {}",
97            self.config.schema_type,
98            self.config.schema_path.display()
99        )
100    }
101}
102
103impl Endpoint for ValidatorEndpoint {
104    fn uri(&self) -> &str {
105        &self.uri
106    }
107
108    fn create_consumer(
109        &self,
110        _rt: Arc<dyn RuntimeObservability>,
111    ) -> Result<Box<dyn Consumer>, CamelError> {
112        Err(CamelError::EndpointCreationFailed(
113            "validator endpoint does not support consumers".to_string(),
114        ))
115    }
116
117    fn create_producer(
118        &self,
119        rt: Arc<dyn RuntimeObservability>,
120        ctx: &ProducerContext,
121    ) -> Result<BoxProcessor, CamelError> {
122        if let Some(ref backend) = self.xsd_backend {
123            backend.set_observability(
124                rt.clone(),
125                ctx.route_id().unwrap_or("validator-bridge").to_string(),
126            );
127        }
128        Ok(BoxProcessor::new(ValidatorProducer {
129            uri: self.uri.clone(),
130            config: self.config.clone(),
131            compiled: Arc::clone(&self.compiled),
132        }))
133    }
134}
135
136#[derive(Clone)]
137struct ValidatorProducer {
138    uri: String,
139    config: ValidatorConfig,
140    compiled: Arc<CompiledValidator>,
141}
142
143impl Service<Exchange> for ValidatorProducer {
144    type Response = Exchange;
145    type Error = CamelError;
146    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
147
148    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
149        Poll::Ready(Ok(()))
150    }
151
152    fn call(&mut self, exchange: Exchange) -> Self::Future {
153        let compiled = Arc::clone(&self.compiled);
154        let uri = self.uri.clone();
155        let config = self.config.clone();
156        Box::pin(async move {
157            debug!(
158                uri = %camel_api::redact::redact_url_fail_closed(&uri),
159                "validating exchange body"
160            );
161
162            // VAL-003: header_name mode — validate header value instead of body
163            if let Some(ref header_name) = config.header_name {
164                match exchange.input.header(header_name) {
165                    Some(value) => {
166                        // Convert header Value to a string body for validation
167                        let header_str = match value.as_str() {
168                            Some(s) => s.to_string(),
169                            None => value.to_string(),
170                        };
171                        let header_body = camel_component_api::Body::Text(header_str);
172                        compiled.validate(&header_body).await?;
173                    }
174                    None => {
175                        // VAL-004: failOnNullHeader
176                        if config.fail_on_null_header {
177                            return Err(CamelError::ProcessorError(format!(
178                                "header '{header_name}' is missing and failOnNullHeader is true"
179                            )));
180                        }
181                        // Pass through — no validation
182                    }
183                }
184                return Ok(exchange);
185            }
186
187            // VAL-002: failOnNullBody
188            if exchange.input.body.is_empty() {
189                if config.fail_on_null_body {
190                    return Err(CamelError::ProcessorError(
191                        "body is empty and failOnNullBody is true".to_string(),
192                    ));
193                }
194                // Pass through — no validation
195                return Ok(exchange);
196            }
197
198            compiled.validate(&exchange.input.body).await?;
199            Ok(exchange)
200        })
201    }
202}
203
204#[cfg(test)]
205mod tests {
206    use camel_component_api::test_support::PanicRuntimeObservability;
207    fn rt() -> std::sync::Arc<dyn camel_component_api::RuntimeObservability> {
208        std::sync::Arc::new(PanicRuntimeObservability)
209    }
210
211    use super::*;
212    use crate::error::ValidatorError;
213    use async_trait::async_trait;
214    use camel_component_api::{Message, NoOpComponentContext};
215    use std::io::Write;
216    use std::sync::atomic::{AtomicUsize, Ordering};
217    use tower::ServiceExt;
218
219    fn json_schema_file() -> tempfile::NamedTempFile {
220        let mut f = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
221        f.write_all(br#"{"type":"object","required":["id"]}"#)
222            .unwrap();
223        f
224    }
225
226    fn xsd_file() -> tempfile::NamedTempFile {
227        let mut f = tempfile::Builder::new().suffix(".xsd").tempfile().unwrap();
228        f.write_all(
229            br#"<?xml version="1.0"?><xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"><xs:element name="order" type="xs:string"/></xs:schema>"#,
230        ).unwrap();
231        f
232    }
233
234    #[derive(Debug)]
235    struct MockXsdBridge {
236        register_calls: AtomicUsize,
237        validate_calls: AtomicUsize,
238        register_error: Option<ValidatorError>,
239        validate_error: Option<ValidatorError>,
240    }
241
242    #[async_trait]
243    impl XsdBridge for MockXsdBridge {
244        async fn register(&self, _xsd_bytes: Vec<u8>) -> Result<String, ValidatorError> {
245            self.register_calls.fetch_add(1, Ordering::SeqCst);
246            if let Some(err) = &self.register_error {
247                return Err(err.clone());
248            }
249            Ok("xsd-mock-id".to_string())
250        }
251
252        async fn validate(
253            &self,
254            _schema_id: &str,
255            _doc_bytes: Vec<u8>,
256        ) -> Result<(), ValidatorError> {
257            self.validate_calls.fetch_add(1, Ordering::SeqCst);
258            if let Some(err) = &self.validate_error {
259                return Err(err.clone());
260            }
261            Ok(())
262        }
263    }
264
265    #[test]
266    fn scheme_is_validator() {
267        assert_eq!(ValidatorComponent::new().scheme(), "validator");
268    }
269
270    #[test]
271    fn consumer_not_supported() {
272        let f = json_schema_file();
273        let uri = format!("validator:{}", f.path().display());
274        let ep = ValidatorComponent::new()
275            .create_endpoint(&uri, &NoOpComponentContext)
276            .unwrap();
277        assert!(ep.create_consumer(rt()).is_err());
278    }
279
280    #[test]
281    fn endpoint_creation_fails_for_missing_schema() {
282        let result = ValidatorComponent::new()
283            .create_endpoint("validator:/nonexistent/schema.json", &NoOpComponentContext);
284        assert!(result.is_err());
285    }
286
287    #[tokio::test]
288    async fn valid_json_body_passes_through() {
289        let f = json_schema_file();
290        let uri = format!("validator:{}", f.path().display());
291        let ep = ValidatorComponent::new()
292            .create_endpoint(&uri, &NoOpComponentContext)
293            .unwrap();
294        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
295        let exchange = Exchange::new(Message::new(camel_component_api::Body::Json(
296            serde_json::json!({"id": "1"}),
297        )));
298        let result = producer.oneshot(exchange).await;
299        assert!(result.is_ok());
300    }
301
302    #[tokio::test]
303    async fn invalid_json_body_returns_err() {
304        let f = json_schema_file();
305        let uri = format!("validator:{}", f.path().display());
306        let ep = ValidatorComponent::new()
307            .create_endpoint(&uri, &NoOpComponentContext)
308            .unwrap();
309        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
310        let exchange = Exchange::new(Message::new(camel_component_api::Body::Json(
311            serde_json::json!({"name": "x"}),
312        )));
313        let result = producer.oneshot(exchange).await;
314        assert!(result.is_err());
315        let msg = result.unwrap_err().to_string();
316        assert!(msg.contains("validation failed"), "got: {msg}");
317    }
318
319    #[tokio::test]
320    async fn valid_xml_body_passes() {
321        let backend = Arc::new(MockXsdBridge {
322            register_calls: AtomicUsize::new(0),
323            validate_calls: AtomicUsize::new(0),
324            register_error: None,
325            validate_error: None,
326        });
327        let f = xsd_file();
328        let uri = format!("validator:{}", f.path().display());
329        let ep = ValidatorComponent::with_xsd_bridge(backend)
330            .create_endpoint(&uri, &NoOpComponentContext)
331            .unwrap();
332        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
333        let exchange = Exchange::new(Message::new(camel_component_api::Body::Xml(
334            "<order>hello</order>".to_string(),
335        )));
336        assert!(producer.oneshot(exchange).await.is_ok());
337    }
338
339    #[tokio::test]
340    async fn xsd_bridge_register_and_validate_mock() {
341        let backend = Arc::new(MockXsdBridge {
342            register_calls: AtomicUsize::new(0),
343            validate_calls: AtomicUsize::new(0),
344            register_error: None,
345            validate_error: None,
346        });
347
348        let f = xsd_file();
349        let uri = format!("validator:{}", f.path().display());
350        let ep = ValidatorComponent::with_xsd_bridge(Arc::clone(&backend) as Arc<dyn XsdBridge>)
351            .create_endpoint(&uri, &NoOpComponentContext)
352            .unwrap();
353
354        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
355        let exchange = Exchange::new(Message::new(camel_component_api::Body::Xml(
356            "<order>ok</order>".to_string(),
357        )));
358        assert!(producer.oneshot(exchange).await.is_ok());
359        assert_eq!(backend.register_calls.load(Ordering::SeqCst), 1);
360        assert_eq!(backend.validate_calls.load(Ordering::SeqCst), 1);
361    }
362
363    #[tokio::test]
364    async fn xsd_bridge_register_error_propagates_on_validate() {
365        let backend = Arc::new(MockXsdBridge {
366            register_calls: AtomicUsize::new(0),
367            validate_calls: AtomicUsize::new(0),
368            register_error: Some(ValidatorError::CompilationFailed {
369                message: "COMPILATION_FAILED".to_string(),
370                source: None,
371            }),
372            validate_error: None,
373        });
374        let f = xsd_file();
375        let uri = format!("validator:{}", f.path().display());
376        // Endpoint creation now always succeeds for XSD (registration is deferred).
377        let ep = ValidatorComponent::with_xsd_bridge(backend)
378            .create_endpoint(&uri, &NoOpComponentContext)
379            .expect("endpoint creation should succeed");
380        // The error surfaces when the first message is processed (register is called).
381        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
382        let exchange = Exchange::new(Message::new(camel_component_api::Body::Xml(
383            "<order/>".to_string(),
384        )));
385        let err = producer
386            .oneshot(exchange)
387            .await
388            .expect_err("expected validate to fail due to registration error");
389        assert!(err.to_string().contains("COMPILATION_FAILED"));
390    }
391
392    #[tokio::test]
393    async fn test_validator_rejects_oversized_payload() {
394        // Build validator with maxPayloadBytes=100
395        let f = json_schema_file();
396        let uri = format!("validator:{}?maxPayloadBytes=100", f.path().display());
397        let ep = ValidatorComponent::new()
398            .create_endpoint(&uri, &NoOpComponentContext)
399            .unwrap();
400        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
401        // Send a body that is definitely > 100 bytes
402        let big_body: String = "x".repeat(200);
403        let exchange = Exchange::new(Message::new(camel_component_api::Body::Text(big_body)));
404        let result = producer.oneshot(exchange).await;
405        assert!(result.is_err(), "expected oversized payload to be rejected");
406        let msg = result.unwrap_err().to_string();
407        assert!(
408            msg.contains("payload too large"),
409            "expected 'payload too large' in error, got: {msg}"
410        );
411    }
412
413    #[tokio::test]
414    async fn test_validator_allows_payload_under_limit() {
415        let f = json_schema_file();
416        let uri = format!("validator:{}?maxPayloadBytes=1024", f.path().display());
417        let ep = ValidatorComponent::new()
418            .create_endpoint(&uri, &NoOpComponentContext)
419            .unwrap();
420        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
421        let exchange = Exchange::new(Message::new(camel_component_api::Body::Json(
422            serde_json::json!({"id": "1"}),
423        )));
424        let result = producer.oneshot(exchange).await;
425        assert!(result.is_ok(), "expected valid payload under limit to pass");
426    }
427
428    #[tokio::test]
429    async fn test_validator_no_limit_allows_any_size() {
430        let f = json_schema_file();
431        let uri = format!("validator:{}", f.path().display());
432        let ep = ValidatorComponent::new()
433            .create_endpoint(&uri, &NoOpComponentContext)
434            .unwrap();
435        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
436        // Valid JSON that would exceed a 10-byte limit
437        let exchange = Exchange::new(Message::new(camel_component_api::Body::Json(
438            serde_json::json!({"id": "this is a longer value that would exceed small limits"}),
439        )));
440        let result = producer.oneshot(exchange).await;
441        assert!(
442            result.is_ok(),
443            "expected no-limit validator to pass any valid payload"
444        );
445    }
446
447    #[tokio::test]
448    async fn test_fail_on_null_body_default_rejects_empty() {
449        let f = json_schema_file();
450        let uri = format!("validator:{}", f.path().display());
451        let ep = ValidatorComponent::new()
452            .create_endpoint(&uri, &NoOpComponentContext)
453            .unwrap();
454        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
455        let exchange = Exchange::new(Message::new(camel_component_api::Body::Empty));
456        let result = producer.oneshot(exchange).await;
457        assert!(result.is_err());
458        let msg = result.unwrap_err().to_string();
459        assert!(
460            msg.contains("failOnNullBody"),
461            "expected failOnNullBody in error, got: {msg}"
462        );
463    }
464
465    #[tokio::test]
466    async fn test_fail_on_null_body_false_passes_empty() {
467        let f = json_schema_file();
468        let uri = format!("validator:{}?failOnNullBody=false", f.path().display());
469        let ep = ValidatorComponent::new()
470            .create_endpoint(&uri, &NoOpComponentContext)
471            .unwrap();
472        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
473        let exchange = Exchange::new(Message::new(camel_component_api::Body::Empty));
474        let result = producer.oneshot(exchange).await;
475        assert!(
476            result.is_ok(),
477            "expected empty body to pass with failOnNullBody=false"
478        );
479    }
480
481    #[tokio::test]
482    async fn test_header_name_validation_uses_header_value() {
483        let f = json_schema_file();
484        let uri = format!("validator:{}?headerName=X-Data", f.path().display());
485        let ep = ValidatorComponent::new()
486            .create_endpoint(&uri, &NoOpComponentContext)
487            .unwrap();
488        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
489        let mut msg = Message::new(camel_component_api::Body::Empty);
490        msg.set_header("X-Data", serde_json::json!({"id": "1"}).to_string());
491        let exchange = Exchange::new(msg);
492        let result = producer.oneshot(exchange).await;
493        // Header value {"id":"1"} is valid JSON matching schema
494        assert!(
495            result.is_ok(),
496            "expected valid header to pass: {:?}",
497            result
498        );
499    }
500
501    #[tokio::test]
502    async fn test_header_name_missing_header_fails_by_default() {
503        let f = json_schema_file();
504        let uri = format!("validator:{}?headerName=X-Missing", f.path().display());
505        let ep = ValidatorComponent::new()
506            .create_endpoint(&uri, &NoOpComponentContext)
507            .unwrap();
508        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
509        let exchange = Exchange::new(Message::new(camel_component_api::Body::Empty));
510        let result = producer.oneshot(exchange).await;
511        assert!(result.is_err());
512        let msg = result.unwrap_err().to_string();
513        assert!(
514            msg.contains("X-Missing"),
515            "expected header name in error, got: {msg}"
516        );
517    }
518
519    #[tokio::test]
520    async fn test_fail_on_null_header_false_passes_missing_header() {
521        let f = json_schema_file();
522        let uri = format!(
523            "validator:{}?headerName=X-Missing&failOnNullHeader=false",
524            f.path().display()
525        );
526        let ep = ValidatorComponent::new()
527            .create_endpoint(&uri, &NoOpComponentContext)
528            .unwrap();
529        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
530        let exchange = Exchange::new(Message::new(camel_component_api::Body::Empty));
531        let result = producer.oneshot(exchange).await;
532        assert!(
533            result.is_ok(),
534            "expected missing header to pass with failOnNullHeader=false"
535        );
536    }
537
538    #[test]
539    fn test_relaxng_schema_type_rejected_at_creation() {
540        let mut f = tempfile::Builder::new().suffix(".rng").tempfile().unwrap();
541        use std::io::Write;
542        f.write_all(b"<grammar/>").unwrap();
543        let uri = format!("validator:{}", f.path().display());
544        let result = ValidatorComponent::new().create_endpoint(&uri, &NoOpComponentContext);
545        let err = result.err().expect("expected endpoint creation to fail");
546        let msg = err.to_string();
547        assert!(
548            msg.contains("not yet supported"),
549            "expected 'not yet supported' in error, got: {msg}"
550        );
551    }
552
553    #[test]
554    fn test_schematron_schema_type_rejected_at_creation() {
555        let mut f = tempfile::Builder::new().suffix(".sch").tempfile().unwrap();
556        use std::io::Write;
557        f.write_all(b"<schema/>").unwrap();
558        let uri = format!("validator:{}", f.path().display());
559        let result = ValidatorComponent::new().create_endpoint(&uri, &NoOpComponentContext);
560        let err = result.err().expect("expected endpoint creation to fail");
561        let msg = err.to_string();
562        assert!(
563            msg.contains("not yet supported"),
564            "expected 'not yet supported' in error, got: {msg}"
565        );
566    }
567
568    #[test]
569    fn test_schema_info_returns_description() {
570        let f = json_schema_file();
571        let uri = format!("validator:{}", f.path().display());
572        let ep = ValidatorComponent::new()
573            .create_endpoint(&uri, &NoOpComponentContext)
574            .unwrap();
575        // The endpoint is a Box<dyn Endpoint>, so we can't directly call schema_info().
576        // We verify endpoint creation succeeds (schema compiled).
577        assert_eq!(ep.uri(), uri);
578    }
579}
camel_component_validator/component.rs

camel_component_validator/
component.rs
