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!(uri = uri, "validating exchange body");
158
159            // VAL-003: header_name mode — validate header value instead of body
160            if let Some(ref header_name) = config.header_name {
161                match exchange.input.header(header_name) {
162                    Some(value) => {
163                        // Convert header Value to a string body for validation
164                        let header_str = match value.as_str() {
165                            Some(s) => s.to_string(),
166                            None => value.to_string(),
167                        };
168                        let header_body = camel_component_api::Body::Text(header_str);
169                        compiled.validate(&header_body).await?;
170                    }
171                    None => {
172                        // VAL-004: failOnNullHeader
173                        if config.fail_on_null_header {
174                            return Err(CamelError::ProcessorError(format!(
175                                "header '{header_name}' is missing and failOnNullHeader is true"
176                            )));
177                        }
178                        // Pass through — no validation
179                    }
180                }
181                return Ok(exchange);
182            }
183
184            // VAL-002: failOnNullBody
185            if exchange.input.body.is_empty() {
186                if config.fail_on_null_body {
187                    return Err(CamelError::ProcessorError(
188                        "body is empty and failOnNullBody is true".to_string(),
189                    ));
190                }
191                // Pass through — no validation
192                return Ok(exchange);
193            }
194
195            compiled.validate(&exchange.input.body).await?;
196            Ok(exchange)
197        })
198    }
199}
200
201#[cfg(test)]
202mod tests {
203    use camel_component_api::test_support::PanicRuntimeObservability;
204    fn rt() -> std::sync::Arc<dyn camel_component_api::RuntimeObservability> {
205        std::sync::Arc::new(PanicRuntimeObservability)
206    }
207
208    use super::*;
209    use crate::error::ValidatorError;
210    use async_trait::async_trait;
211    use camel_component_api::{Message, NoOpComponentContext};
212    use std::io::Write;
213    use std::sync::atomic::{AtomicUsize, Ordering};
214    use tower::ServiceExt;
215
216    fn json_schema_file() -> tempfile::NamedTempFile {
217        let mut f = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
218        f.write_all(br#"{"type":"object","required":["id"]}"#)
219            .unwrap();
220        f
221    }
222
223    fn xsd_file() -> tempfile::NamedTempFile {
224        let mut f = tempfile::Builder::new().suffix(".xsd").tempfile().unwrap();
225        f.write_all(
226            br#"<?xml version="1.0"?><xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"><xs:element name="order" type="xs:string"/></xs:schema>"#,
227        ).unwrap();
228        f
229    }
230
231    #[derive(Debug)]
232    struct MockXsdBridge {
233        register_calls: AtomicUsize,
234        validate_calls: AtomicUsize,
235        register_error: Option<ValidatorError>,
236        validate_error: Option<ValidatorError>,
237    }
238
239    #[async_trait]
240    impl XsdBridge for MockXsdBridge {
241        async fn register(&self, _xsd_bytes: Vec<u8>) -> Result<String, ValidatorError> {
242            self.register_calls.fetch_add(1, Ordering::SeqCst);
243            if let Some(err) = &self.register_error {
244                return Err(err.clone());
245            }
246            Ok("xsd-mock-id".to_string())
247        }
248
249        async fn validate(
250            &self,
251            _schema_id: &str,
252            _doc_bytes: Vec<u8>,
253        ) -> Result<(), ValidatorError> {
254            self.validate_calls.fetch_add(1, Ordering::SeqCst);
255            if let Some(err) = &self.validate_error {
256                return Err(err.clone());
257            }
258            Ok(())
259        }
260    }
261
262    #[test]
263    fn scheme_is_validator() {
264        assert_eq!(ValidatorComponent::new().scheme(), "validator");
265    }
266
267    #[test]
268    fn consumer_not_supported() {
269        let f = json_schema_file();
270        let uri = format!("validator:{}", f.path().display());
271        let ep = ValidatorComponent::new()
272            .create_endpoint(&uri, &NoOpComponentContext)
273            .unwrap();
274        assert!(ep.create_consumer(rt()).is_err());
275    }
276
277    #[test]
278    fn endpoint_creation_fails_for_missing_schema() {
279        let result = ValidatorComponent::new()
280            .create_endpoint("validator:/nonexistent/schema.json", &NoOpComponentContext);
281        assert!(result.is_err());
282    }
283
284    #[tokio::test]
285    async fn valid_json_body_passes_through() {
286        let f = json_schema_file();
287        let uri = format!("validator:{}", f.path().display());
288        let ep = ValidatorComponent::new()
289            .create_endpoint(&uri, &NoOpComponentContext)
290            .unwrap();
291        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
292        let exchange = Exchange::new(Message::new(camel_component_api::Body::Json(
293            serde_json::json!({"id": "1"}),
294        )));
295        let result = producer.oneshot(exchange).await;
296        assert!(result.is_ok());
297    }
298
299    #[tokio::test]
300    async fn invalid_json_body_returns_err() {
301        let f = json_schema_file();
302        let uri = format!("validator:{}", f.path().display());
303        let ep = ValidatorComponent::new()
304            .create_endpoint(&uri, &NoOpComponentContext)
305            .unwrap();
306        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
307        let exchange = Exchange::new(Message::new(camel_component_api::Body::Json(
308            serde_json::json!({"name": "x"}),
309        )));
310        let result = producer.oneshot(exchange).await;
311        assert!(result.is_err());
312        let msg = result.unwrap_err().to_string();
313        assert!(msg.contains("validation failed"), "got: {msg}");
314    }
315
316    #[tokio::test]
317    async fn valid_xml_body_passes() {
318        let backend = Arc::new(MockXsdBridge {
319            register_calls: AtomicUsize::new(0),
320            validate_calls: AtomicUsize::new(0),
321            register_error: None,
322            validate_error: None,
323        });
324        let f = xsd_file();
325        let uri = format!("validator:{}", f.path().display());
326        let ep = ValidatorComponent::with_xsd_bridge(backend)
327            .create_endpoint(&uri, &NoOpComponentContext)
328            .unwrap();
329        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
330        let exchange = Exchange::new(Message::new(camel_component_api::Body::Xml(
331            "<order>hello</order>".to_string(),
332        )));
333        assert!(producer.oneshot(exchange).await.is_ok());
334    }
335
336    #[tokio::test]
337    async fn xsd_bridge_register_and_validate_mock() {
338        let backend = Arc::new(MockXsdBridge {
339            register_calls: AtomicUsize::new(0),
340            validate_calls: AtomicUsize::new(0),
341            register_error: None,
342            validate_error: None,
343        });
344
345        let f = xsd_file();
346        let uri = format!("validator:{}", f.path().display());
347        let ep = ValidatorComponent::with_xsd_bridge(Arc::clone(&backend) as Arc<dyn XsdBridge>)
348            .create_endpoint(&uri, &NoOpComponentContext)
349            .unwrap();
350
351        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
352        let exchange = Exchange::new(Message::new(camel_component_api::Body::Xml(
353            "<order>ok</order>".to_string(),
354        )));
355        assert!(producer.oneshot(exchange).await.is_ok());
356        assert_eq!(backend.register_calls.load(Ordering::SeqCst), 1);
357        assert_eq!(backend.validate_calls.load(Ordering::SeqCst), 1);
358    }
359
360    #[tokio::test]
361    async fn xsd_bridge_register_error_propagates_on_validate() {
362        let backend = Arc::new(MockXsdBridge {
363            register_calls: AtomicUsize::new(0),
364            validate_calls: AtomicUsize::new(0),
365            register_error: Some(ValidatorError::CompilationFailed {
366                message: "COMPILATION_FAILED".to_string(),
367                source: None,
368            }),
369            validate_error: None,
370        });
371        let f = xsd_file();
372        let uri = format!("validator:{}", f.path().display());
373        // Endpoint creation now always succeeds for XSD (registration is deferred).
374        let ep = ValidatorComponent::with_xsd_bridge(backend)
375            .create_endpoint(&uri, &NoOpComponentContext)
376            .expect("endpoint creation should succeed");
377        // The error surfaces when the first message is processed (register is called).
378        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
379        let exchange = Exchange::new(Message::new(camel_component_api::Body::Xml(
380            "<order/>".to_string(),
381        )));
382        let err = producer
383            .oneshot(exchange)
384            .await
385            .expect_err("expected validate to fail due to registration error");
386        assert!(err.to_string().contains("COMPILATION_FAILED"));
387    }
388
389    #[tokio::test]
390    async fn test_validator_rejects_oversized_payload() {
391        // Build validator with maxPayloadBytes=100
392        let f = json_schema_file();
393        let uri = format!("validator:{}?maxPayloadBytes=100", f.path().display());
394        let ep = ValidatorComponent::new()
395            .create_endpoint(&uri, &NoOpComponentContext)
396            .unwrap();
397        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
398        // Send a body that is definitely > 100 bytes
399        let big_body: String = "x".repeat(200);
400        let exchange = Exchange::new(Message::new(camel_component_api::Body::Text(big_body)));
401        let result = producer.oneshot(exchange).await;
402        assert!(result.is_err(), "expected oversized payload to be rejected");
403        let msg = result.unwrap_err().to_string();
404        assert!(
405            msg.contains("payload too large"),
406            "expected 'payload too large' in error, got: {msg}"
407        );
408    }
409
410    #[tokio::test]
411    async fn test_validator_allows_payload_under_limit() {
412        let f = json_schema_file();
413        let uri = format!("validator:{}?maxPayloadBytes=1024", f.path().display());
414        let ep = ValidatorComponent::new()
415            .create_endpoint(&uri, &NoOpComponentContext)
416            .unwrap();
417        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
418        let exchange = Exchange::new(Message::new(camel_component_api::Body::Json(
419            serde_json::json!({"id": "1"}),
420        )));
421        let result = producer.oneshot(exchange).await;
422        assert!(result.is_ok(), "expected valid payload under limit to pass");
423    }
424
425    #[tokio::test]
426    async fn test_validator_no_limit_allows_any_size() {
427        let f = json_schema_file();
428        let uri = format!("validator:{}", f.path().display());
429        let ep = ValidatorComponent::new()
430            .create_endpoint(&uri, &NoOpComponentContext)
431            .unwrap();
432        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
433        // Valid JSON that would exceed a 10-byte limit
434        let exchange = Exchange::new(Message::new(camel_component_api::Body::Json(
435            serde_json::json!({"id": "this is a longer value that would exceed small limits"}),
436        )));
437        let result = producer.oneshot(exchange).await;
438        assert!(
439            result.is_ok(),
440            "expected no-limit validator to pass any valid payload"
441        );
442    }
443
444    #[tokio::test]
445    async fn test_fail_on_null_body_default_rejects_empty() {
446        let f = json_schema_file();
447        let uri = format!("validator:{}", f.path().display());
448        let ep = ValidatorComponent::new()
449            .create_endpoint(&uri, &NoOpComponentContext)
450            .unwrap();
451        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
452        let exchange = Exchange::new(Message::new(camel_component_api::Body::Empty));
453        let result = producer.oneshot(exchange).await;
454        assert!(result.is_err());
455        let msg = result.unwrap_err().to_string();
456        assert!(
457            msg.contains("failOnNullBody"),
458            "expected failOnNullBody in error, got: {msg}"
459        );
460    }
461
462    #[tokio::test]
463    async fn test_fail_on_null_body_false_passes_empty() {
464        let f = json_schema_file();
465        let uri = format!("validator:{}?failOnNullBody=false", f.path().display());
466        let ep = ValidatorComponent::new()
467            .create_endpoint(&uri, &NoOpComponentContext)
468            .unwrap();
469        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
470        let exchange = Exchange::new(Message::new(camel_component_api::Body::Empty));
471        let result = producer.oneshot(exchange).await;
472        assert!(
473            result.is_ok(),
474            "expected empty body to pass with failOnNullBody=false"
475        );
476    }
477
478    #[tokio::test]
479    async fn test_header_name_validation_uses_header_value() {
480        let f = json_schema_file();
481        let uri = format!("validator:{}?headerName=X-Data", f.path().display());
482        let ep = ValidatorComponent::new()
483            .create_endpoint(&uri, &NoOpComponentContext)
484            .unwrap();
485        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
486        let mut msg = Message::new(camel_component_api::Body::Empty);
487        msg.set_header("X-Data", serde_json::json!({"id": "1"}).to_string());
488        let exchange = Exchange::new(msg);
489        let result = producer.oneshot(exchange).await;
490        // Header value {"id":"1"} is valid JSON matching schema
491        assert!(
492            result.is_ok(),
493            "expected valid header to pass: {:?}",
494            result
495        );
496    }
497
498    #[tokio::test]
499    async fn test_header_name_missing_header_fails_by_default() {
500        let f = json_schema_file();
501        let uri = format!("validator:{}?headerName=X-Missing", f.path().display());
502        let ep = ValidatorComponent::new()
503            .create_endpoint(&uri, &NoOpComponentContext)
504            .unwrap();
505        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
506        let exchange = Exchange::new(Message::new(camel_component_api::Body::Empty));
507        let result = producer.oneshot(exchange).await;
508        assert!(result.is_err());
509        let msg = result.unwrap_err().to_string();
510        assert!(
511            msg.contains("X-Missing"),
512            "expected header name in error, got: {msg}"
513        );
514    }
515
516    #[tokio::test]
517    async fn test_fail_on_null_header_false_passes_missing_header() {
518        let f = json_schema_file();
519        let uri = format!(
520            "validator:{}?headerName=X-Missing&failOnNullHeader=false",
521            f.path().display()
522        );
523        let ep = ValidatorComponent::new()
524            .create_endpoint(&uri, &NoOpComponentContext)
525            .unwrap();
526        let producer = ep.create_producer(rt(), &ProducerContext::new()).unwrap();
527        let exchange = Exchange::new(Message::new(camel_component_api::Body::Empty));
528        let result = producer.oneshot(exchange).await;
529        assert!(
530            result.is_ok(),
531            "expected missing header to pass with failOnNullHeader=false"
532        );
533    }
534
535    #[test]
536    fn test_relaxng_schema_type_rejected_at_creation() {
537        let mut f = tempfile::Builder::new().suffix(".rng").tempfile().unwrap();
538        use std::io::Write;
539        f.write_all(b"<grammar/>").unwrap();
540        let uri = format!("validator:{}", f.path().display());
541        let result = ValidatorComponent::new().create_endpoint(&uri, &NoOpComponentContext);
542        let err = result.err().expect("expected endpoint creation to fail");
543        let msg = err.to_string();
544        assert!(
545            msg.contains("not yet supported"),
546            "expected 'not yet supported' in error, got: {msg}"
547        );
548    }
549
550    #[test]
551    fn test_schematron_schema_type_rejected_at_creation() {
552        let mut f = tempfile::Builder::new().suffix(".sch").tempfile().unwrap();
553        use std::io::Write;
554        f.write_all(b"<schema/>").unwrap();
555        let uri = format!("validator:{}", f.path().display());
556        let result = ValidatorComponent::new().create_endpoint(&uri, &NoOpComponentContext);
557        let err = result.err().expect("expected endpoint creation to fail");
558        let msg = err.to_string();
559        assert!(
560            msg.contains("not yet supported"),
561            "expected 'not yet supported' in error, got: {msg}"
562        );
563    }
564
565    #[test]
566    fn test_schema_info_returns_description() {
567        let f = json_schema_file();
568        let uri = format!("validator:{}", f.path().display());
569        let ep = ValidatorComponent::new()
570            .create_endpoint(&uri, &NoOpComponentContext)
571            .unwrap();
572        // The endpoint is a Box<dyn Endpoint>, so we can't directly call schema_info().
573        // We verify endpoint creation succeeds (schema compiled).
574        assert_eq!(ep.uri(), uri);
575    }
576}
camel_component_validator/component.rs

camel_component_validator/
component.rs
