1use async_trait::async_trait;
7use camel_language_api::{Body, Exchange, Value};
8use camel_language_api::{Expression, Language, LanguageError, Predicate};
9use jsonpath_rust::parser::model::JpQuery;
10use jsonpath_rust::parser::parse_json_path;
11use jsonpath_rust::query::js_path_process;
12use serde_json::Value as JsonValue;
13#[cfg(test)]
14thread_local! {
15 static COMPILE_COUNT: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
16}
17
18const DEFAULT_MAX_DEPTH: usize = 64;
20
21const DEFAULT_MAX_INPUT_BYTES: usize = 16 * 1024 * 1024; #[derive(Debug, Clone)]
27pub struct JsonPathConfig {
28 pub max_input_bytes: Option<usize>,
31 pub max_depth: Option<usize>,
34}
35
36impl JsonPathConfig {
37 fn effective_max_depth(&self) -> usize {
39 self.max_depth.unwrap_or(DEFAULT_MAX_DEPTH)
40 }
41}
42
43impl Default for JsonPathConfig {
44 fn default() -> Self {
45 Self {
46 max_input_bytes: Some(DEFAULT_MAX_INPUT_BYTES),
47 max_depth: None,
48 }
49 }
50}
51
52pub struct JsonPathLanguage {
54 config: JsonPathConfig,
55}
56
57impl JsonPathLanguage {
58 pub fn new() -> Self {
60 Self {
61 config: JsonPathConfig::default(),
62 }
63 }
64
65 pub fn with_config(config: JsonPathConfig) -> Self {
67 Self { config }
68 }
69}
70
71impl Default for JsonPathLanguage {
72 fn default() -> Self {
73 Self::new()
74 }
75}
76
77struct JsonPathExpression {
78 query: JpQuery,
79 config: JsonPathConfig,
80}
81
82struct JsonPathPredicate {
83 query: JpQuery,
84 config: JsonPathConfig,
85}
86
87fn check_depth(value: &JsonValue, max_depth: usize) -> Result<(), LanguageError> {
89 fn recurse(value: &JsonValue, max_depth: usize, current: usize) -> Result<(), LanguageError> {
90 if current > max_depth {
91 return Err(LanguageError::EvalError(format!(
92 "JSON nesting depth {current} exceeds limit of {max_depth}"
93 )));
94 }
95 match value {
96 JsonValue::Object(map) => {
97 for v in map.values() {
98 recurse(v, max_depth, current + 1)?;
99 }
100 Ok(())
101 }
102 JsonValue::Array(arr) => {
103 for v in arr {
104 recurse(v, max_depth, current + 1)?;
105 }
106 Ok(())
107 }
108 _ => Ok(()),
109 }
110 }
111 recurse(value, max_depth, 0)
112}
113
114fn extract_json(exchange: &Exchange, config: &JsonPathConfig) -> Result<JsonValue, LanguageError> {
120 let max_depth = config.effective_max_depth();
121
122 match &exchange.input.body {
123 Body::Json(v) => {
124 check_depth(v, max_depth)?;
126 Ok(v.clone())
127 }
128 Body::Text(s) => {
129 if let Some(limit) = config.max_input_bytes
131 && s.len() > limit
132 {
133 return Err(LanguageError::EvalError(format!(
134 "input size {} bytes exceeds limit of {limit} bytes",
135 s.len()
136 )));
137 }
138 let value: JsonValue = serde_json::from_str(s)
139 .map_err(|e| LanguageError::EvalError(format!("body is not valid JSON: {e}")))?;
140 check_depth(&value, max_depth)?;
141 Ok(value)
142 }
143 other => other
144 .clone()
145 .try_into_json()
146 .map_err(|e| {
147 LanguageError::EvalError(format!("body is not JSON and cannot be coerced: {e}"))
148 })
149 .and_then(|b| match b {
150 Body::Json(v) => {
151 check_depth(&v, max_depth)?;
152 Ok(v)
153 }
154 _ => Err(LanguageError::EvalError(
155 "body coercion did not produce JSON".into(),
156 )),
157 }),
158 }
159}
160
161fn run_query(query: &JpQuery, json: &JsonValue) -> Result<JsonValue, LanguageError> {
162 let results = js_path_process(query, json)
163 .map_err(|e| LanguageError::EvalError(format!("jsonpath query '{query}' failed: {e}")))?;
164 let values: Vec<&JsonValue> = results.into_iter().map(|r| r.val).collect();
165 Ok(match values.len() {
166 0 => JsonValue::Null,
167 1 => values[0].clone(),
168 _ => JsonValue::Array(values.into_iter().cloned().collect()),
169 })
170}
171
172#[async_trait]
173impl Expression for JsonPathExpression {
174 async fn evaluate(&self, exchange: &Exchange) -> Result<Value, LanguageError> {
179 let json = extract_json(exchange, &self.config)?;
180 run_query(&self.query, &json)
181 }
182}
183
184#[async_trait]
185impl Predicate for JsonPathPredicate {
186 async fn matches(&self, exchange: &Exchange) -> Result<bool, LanguageError> {
187 let json = extract_json(exchange, &self.config)?;
188 let result = run_query(&self.query, &json)?;
189 Ok(is_truthy(&result))
190 }
191}
192
193fn is_truthy(value: &JsonValue) -> bool {
194 match value {
195 JsonValue::Null => false,
196 JsonValue::Bool(b) => *b,
197 JsonValue::Number(n) => {
198 if let Some(v) = n.as_i64() {
199 return v != 0;
200 }
201 if let Some(v) = n.as_u64() {
202 return v != 0;
203 }
204 if let Some(v) = n.as_f64() {
205 return v != 0.0;
206 }
207 true
208 }
209 JsonValue::String(s) => !s.is_empty(),
210 JsonValue::Array(arr) => !arr.is_empty(),
211 JsonValue::Object(_) => true,
212 }
213}
214
215impl Language for JsonPathLanguage {
216 fn name(&self) -> &'static str {
217 "jsonpath"
218 }
219
220 fn create_expression(&self, script: &str) -> Result<Box<dyn Expression>, LanguageError> {
221 if !script.starts_with('$') {
222 return Err(LanguageError::ParseError {
223 expr: script.to_string(),
224 reason: "JsonPath expression must start with '$'".into(),
225 });
226 }
227 let parsed = parse_json_path(script).map_err(|e| LanguageError::ParseError {
228 expr: script.to_string(),
229 reason: e.to_string(),
230 })?;
231 #[cfg(test)]
232 {
233 COMPILE_COUNT.with(|c| c.set(c.get() + 1));
234 }
235 Ok(Box::new(JsonPathExpression {
236 query: parsed,
237 config: self.config.clone(),
238 }))
239 }
240
241 fn create_predicate(&self, script: &str) -> Result<Box<dyn Predicate>, LanguageError> {
242 if !script.starts_with('$') {
243 return Err(LanguageError::ParseError {
244 expr: script.to_string(),
245 reason: "JsonPath expression must start with '$'".into(),
246 });
247 }
248 let parsed = parse_json_path(script).map_err(|e| LanguageError::ParseError {
249 expr: script.to_string(),
250 reason: e.to_string(),
251 })?;
252 #[cfg(test)]
253 {
254 COMPILE_COUNT.with(|c| c.set(c.get() + 1));
255 }
256 Ok(Box::new(JsonPathPredicate {
257 query: parsed,
258 config: self.config.clone(),
259 }))
260 }
261}
262
263#[cfg(test)]
264mod tests {
265 use super::*;
266 use camel_language_api::Message;
267
268 async fn exchange_with_json(json: &str) -> Exchange {
269 let value: JsonValue = serde_json::from_str(json).unwrap();
270 Exchange::new(Message::new(Body::Json(value)))
271 }
272
273 async fn exchange_with_text_body(text: &str) -> Exchange {
274 Exchange::new(Message::new(Body::Text(text.to_string())))
275 }
276
277 async fn empty_exchange() -> Exchange {
278 Exchange::new(Message::default())
279 }
280
281 async fn default_lang() -> JsonPathLanguage {
282 JsonPathLanguage::new()
283 }
284
285 #[tokio::test]
286 async fn expression_simple_path() {
287 let lang = default_lang().await;
288 let expr = lang.create_expression("$.store.name").unwrap();
289 let ex = exchange_with_json(r#"{"store":{"name":"books"}}"#).await;
290 let result = expr.evaluate(&ex).await.unwrap();
291 assert_eq!(result, JsonValue::String("books".to_string()));
292 }
293
294 #[tokio::test]
295 async fn expression_nested_path() {
296 let lang = default_lang().await;
297 let expr = lang.create_expression("$.a.b.c").unwrap();
298 let ex = exchange_with_json(r#"{"a":{"b":{"c":42}}}"#).await;
299 let result = expr.evaluate(&ex).await.unwrap();
300 assert_eq!(result, JsonValue::Number(42.into()));
301 }
302
303 #[tokio::test]
304 async fn expression_array_index() {
305 let lang = default_lang().await;
306 let expr = lang.create_expression("$.items[0]").unwrap();
307 let ex = exchange_with_json(r#"{"items":["a","b","c"]}"#).await;
308 let result = expr.evaluate(&ex).await.unwrap();
309 assert_eq!(result, JsonValue::String("a".to_string()));
310 }
311
312 #[tokio::test]
313 async fn expression_wildcard() {
314 let lang = default_lang().await;
315 let expr = lang.create_expression("$.items[*].name").unwrap();
316 let ex = exchange_with_json(r#"{"items":[{"name":"a"},{"name":"b"}]}"#).await;
317 let result = expr.evaluate(&ex).await.unwrap();
318 assert_eq!(
319 result,
320 JsonValue::Array(vec![
321 JsonValue::String("a".to_string()),
322 JsonValue::String("b".to_string())
323 ])
324 );
325 }
326
327 #[tokio::test]
328 async fn expression_root_path() {
329 let lang = default_lang().await;
330 let expr = lang.create_expression("$").unwrap();
331 let ex = exchange_with_json(r#"{"x":1}"#).await;
332 let result = expr.evaluate(&ex).await.unwrap();
333 assert_eq!(result["x"], JsonValue::Number(1.into()));
334 }
335
336 #[tokio::test]
337 async fn expression_text_body_with_valid_json() {
338 let lang = default_lang().await;
339 let expr = lang.create_expression("$.name").unwrap();
340 let ex = exchange_with_text_body(r#"{"name":"test"}"#).await;
341 let result = expr.evaluate(&ex).await.unwrap();
342 assert_eq!(result, JsonValue::String("test".to_string()));
343 }
344
345 #[tokio::test]
346 async fn expression_empty_body_is_error() {
347 let lang = default_lang().await;
348 let expr = lang.create_expression("$.x").unwrap();
349 let ex = empty_exchange().await;
350 let result = expr.evaluate(&ex).await;
351 assert!(result.is_err());
352 }
353
354 #[tokio::test]
355 async fn expression_invalid_jsonpath_syntax() {
356 let lang = default_lang().await;
357 let result = lang.create_expression("$[invalid");
358 let err = match result {
359 Err(e) => e,
360 Ok(_) => panic!("expected ParseError"),
361 };
362 match err {
363 LanguageError::ParseError { expr, reason } => {
364 assert!(!expr.is_empty());
365 assert!(!reason.is_empty());
366 }
367 other => panic!("expected ParseError, got {other:?}"),
368 }
369 }
370
371 #[tokio::test]
374 async fn expression_without_dollar_prefix_is_rejected() {
375 let lang = default_lang().await;
376 let result = lang.create_expression("store.name");
377 assert!(result.is_err(), "expected error for missing $ prefix");
378 let err = match result {
379 Err(e) => e,
380 Ok(_) => panic!("expected ParseError"),
381 };
382 match err {
383 LanguageError::ParseError { expr, reason } => {
384 assert_eq!(expr, "store.name");
385 assert!(
386 reason.contains("'$'"),
387 "reason should mention '$', got: {reason}"
388 );
389 }
390 other => panic!("expected ParseError, got {other:?}"),
391 }
392 }
393
394 #[tokio::test]
395 async fn predicate_without_dollar_prefix_is_rejected() {
396 let lang = default_lang().await;
397 let result = lang.create_predicate("store.name");
398 assert!(result.is_err(), "expected error for missing $ prefix");
399 let err = match result {
400 Err(e) => e,
401 Ok(_) => panic!("expected ParseError"),
402 };
403 match err {
404 LanguageError::ParseError { reason, .. } => {
405 assert!(
406 reason.contains("'$'"),
407 "reason should mention '$', got: {reason}"
408 );
409 }
410 other => panic!("expected ParseError, got {other:?}"),
411 }
412 }
413
414 #[tokio::test]
417 async fn expression_deeply_nested_path() {
418 let lang = default_lang().await;
419 let expr = lang.create_expression("$.a.b.c.d").unwrap();
420 let ex = exchange_with_json(r#"{"a":{"b":{"c":{"d":"deep"}}}}"#).await;
421 let result = expr.evaluate(&ex).await.unwrap();
422 assert_eq!(result, JsonValue::String("deep".to_string()));
423 }
424
425 #[tokio::test]
426 async fn expression_array_index_nested() {
427 let lang = default_lang().await;
428 let expr = lang.create_expression("$.data.items[1].name").unwrap();
429 let ex = exchange_with_json(
430 r#"{"data":{"items":[{"name":"first"},{"name":"second"},{"name":"third"}]}}"#,
431 )
432 .await;
433 let result = expr.evaluate(&ex).await.unwrap();
434 assert_eq!(result, JsonValue::String("second".to_string()));
435 }
436
437 #[tokio::test]
438 async fn predicate_non_empty_array_is_true() {
439 let lang = default_lang().await;
440 let pred = lang.create_predicate("$.items[*]").unwrap();
441 let ex = exchange_with_json(r#"{"items":[1,2,3]}"#).await;
442 assert!(pred.matches(&ex).await.unwrap());
443 }
444
445 #[tokio::test]
446 async fn predicate_empty_result_is_false() {
447 let lang = default_lang().await;
448 let pred = lang.create_predicate("$.missing").unwrap();
449 let ex = exchange_with_json(r#"{"other":1}"#).await;
450 assert!(!pred.matches(&ex).await.unwrap());
451 }
452
453 #[tokio::test]
454 async fn predicate_boolean_true() {
455 let lang = default_lang().await;
456 let pred = lang.create_predicate("$.active").unwrap();
457 let ex = exchange_with_json(r#"{"active":true}"#).await;
458 assert!(pred.matches(&ex).await.unwrap());
459 }
460
461 #[tokio::test]
462 async fn predicate_boolean_false() {
463 let lang = default_lang().await;
464 let pred = lang.create_predicate("$.active").unwrap();
465 let ex = exchange_with_json(r#"{"active":false}"#).await;
466 assert!(!pred.matches(&ex).await.unwrap());
467 }
468
469 #[tokio::test]
470 async fn predicate_found_value_is_true() {
471 let lang = default_lang().await;
472 let pred = lang.create_predicate("$.name").unwrap();
473 let ex = exchange_with_json(r#"{"name":"test"}"#).await;
474 assert!(pred.matches(&ex).await.unwrap());
475 }
476
477 #[tokio::test]
478 async fn predicate_zero_is_false() {
479 let lang = default_lang().await;
480 let pred = lang.create_predicate("$.val").unwrap();
481 let ex = exchange_with_json(r#"{"val":0}"#).await;
482 assert!(!pred.matches(&ex).await.unwrap());
483 }
484
485 #[tokio::test]
486 async fn predicate_non_zero_is_true() {
487 let lang = default_lang().await;
488 let pred = lang.create_predicate("$.val").unwrap();
489 let ex = exchange_with_json(r#"{"val":1}"#).await;
490 assert!(pred.matches(&ex).await.unwrap());
491 }
492
493 #[tokio::test]
494 async fn predicate_empty_string_is_false() {
495 let lang = default_lang().await;
496 let pred = lang.create_predicate("$.val").unwrap();
497 let ex = exchange_with_json(r#"{"val":""}"#).await;
498 assert!(!pred.matches(&ex).await.unwrap());
499 }
500
501 #[tokio::test]
502 async fn predicate_non_empty_string_is_true() {
503 let lang = default_lang().await;
504 let pred = lang.create_predicate("$.val").unwrap();
505 let ex = exchange_with_json(r#"{"val":"x"}"#).await;
506 assert!(pred.matches(&ex).await.unwrap());
507 }
508
509 #[tokio::test]
512 async fn oversized_input_is_rejected() {
513 let config = JsonPathConfig {
514 max_input_bytes: Some(100),
515 ..Default::default()
516 };
517 let lang = JsonPathLanguage::with_config(config);
518 let expr = lang.create_expression("$.key").unwrap();
519 let big_value = "x".repeat(200);
521 let big_json = format!(r#"{{"key":"{}"}}"#, big_value);
522 assert!(big_json.len() > 100);
523 let ex = exchange_with_text_body(&big_json).await;
524 let result = expr.evaluate(&ex).await;
525 assert!(
526 result.is_err(),
527 "expected error for oversized input, got {result:?}"
528 );
529 }
530
531 #[tokio::test]
532 async fn input_under_limit_is_accepted() {
533 let config = JsonPathConfig {
534 max_input_bytes: Some(1024),
535 ..Default::default()
536 };
537 let lang = JsonPathLanguage::with_config(config);
538 let expr = lang.create_expression("$.key").unwrap();
539 let ex = exchange_with_text_body(r#"{"key":"value"}"#).await;
540 let result = expr.evaluate(&ex).await;
541 assert!(
542 result.is_ok(),
543 "expected success for input under limit, got {result:?}"
544 );
545 }
546
547 #[tokio::test]
548 async fn deeply_nested_input_is_rejected() {
549 let config = JsonPathConfig {
550 max_depth: Some(5),
551 ..Default::default()
552 };
553 let lang = JsonPathLanguage::with_config(config);
554 let expr = lang.create_expression("$.a").unwrap();
555 let mut json = "1".to_string();
557 for _ in 0..10 {
558 json = format!(r#"{{"a":{json}}}"#);
559 }
560 let ex = exchange_with_text_body(&json).await;
561 let result = expr.evaluate(&ex).await;
562 assert!(
563 result.is_err(),
564 "expected error for deeply nested input, got {result:?}"
565 );
566 }
567
568 #[tokio::test]
569 async fn nesting_within_depth_limit_is_accepted() {
570 let config = JsonPathConfig {
571 max_depth: Some(10),
572 ..Default::default()
573 };
574 let lang = JsonPathLanguage::with_config(config);
575 let expr = lang.create_expression("$.a").unwrap();
576 let mut json = "1".to_string();
578 for _ in 0..5 {
579 json = format!(r#"{{"a":{json}}}"#);
580 }
581 let ex = exchange_with_text_body(&json).await;
582 let result = expr.evaluate(&ex).await;
583 assert!(
584 result.is_ok(),
585 "expected success for nesting within limit, got {result:?}"
586 );
587 }
588
589 #[tokio::test]
590 async fn default_config_has_safe_defaults() {
591 let config = JsonPathConfig::default();
593 assert_eq!(config.max_input_bytes, Some(16 * 1024 * 1024));
594 assert_eq!(config.max_depth, None);
596 }
597
598 #[tokio::test]
599 async fn default_config_rejects_oversized_text_body() {
600 let lang = JsonPathLanguage::new(); let expr = lang.create_expression("$.key").unwrap();
603 let big_value = "x".repeat(16 * 1024 * 1024 + 10);
604 let big_json = format!(r#"{{"key":"{}"}}"#, big_value);
605 let ex = exchange_with_text_body(&big_json).await;
606 let result = expr.evaluate(&ex).await;
607 assert!(
608 result.is_err(),
609 "default config must reject oversized input, got {result:?}"
610 );
611 }
612
613 #[tokio::test]
614 async fn oversized_input_also_rejected_for_predicate() {
615 let config = JsonPathConfig {
616 max_input_bytes: Some(100),
617 ..Default::default()
618 };
619 let lang = JsonPathLanguage::with_config(config);
620 let pred = lang.create_predicate("$.key").unwrap();
621 let big_value = "x".repeat(200);
622 let big_json = format!(r#"{{"key":"{}"}}"#, big_value);
623 let ex = exchange_with_text_body(&big_json).await;
624 let result = pred.matches(&ex).await;
625 assert!(
626 result.is_err(),
627 "expected error for oversized input in predicate, got {result:?}"
628 );
629 }
630
631 #[tokio::test]
632 async fn deeply_nested_input_rejected_for_predicate() {
633 let config = JsonPathConfig {
634 max_depth: Some(3),
635 ..Default::default()
636 };
637 let lang = JsonPathLanguage::with_config(config);
638 let pred = lang.create_predicate("$.a").unwrap();
639 let mut json = "1".to_string();
640 for _ in 0..5 {
641 json = format!(r#"{{"a":{json}}}"#);
642 }
643 let ex = exchange_with_text_body(&json).await;
644 let result = pred.matches(&ex).await;
645 assert!(
646 result.is_err(),
647 "expected error for deeply nested input in predicate, got {result:?}"
648 );
649 }
650
651 #[tokio::test]
652 async fn body_json_no_input_size_check_but_depth_checked() {
653 let config = JsonPathConfig {
656 max_input_bytes: Some(10), max_depth: Some(3),
658 };
659 let lang = JsonPathLanguage::with_config(config);
660 let expr = lang.create_expression("$.a").unwrap();
661 let mut json_str = "1".to_string();
663 for _ in 0..5 {
664 json_str = format!(r#"{{"a":{json_str}}}"#);
665 }
666 let ex = exchange_with_json(&json_str).await;
667 let result = expr.evaluate(&ex).await;
668 assert!(
669 result.is_err(),
670 "expected depth error for pre-parsed JSON, got {result:?}"
671 );
672 }
673
674 #[test]
680 fn compile_once_types_are_send_sync() {
681 fn assert_send_sync<T: Send + Sync>() {}
682 assert_send_sync::<JsonPathExpression>();
683 assert_send_sync::<JsonPathPredicate>();
684 assert_send_sync::<JpQuery>();
685 }
686
687 #[tokio::test]
696 async fn compilation_happens_only_once_for_expression() {
697 let before = COMPILE_COUNT.with(std::cell::Cell::get);
698 let lang = JsonPathLanguage::new();
699 let expr = lang.create_expression("$.foo.bar").unwrap();
700 let after_create = COMPILE_COUNT.with(std::cell::Cell::get);
701 assert_eq!(
702 after_create,
703 before + 1,
704 "create_expression must compile exactly once"
705 );
706
707 for _ in 0..50 {
709 let ex = exchange_with_json(r#"{"foo":{"bar":"baz"}}"#).await;
710 let result = expr.evaluate(&ex).await.unwrap();
711 assert_eq!(result, JsonValue::String("baz".to_string()));
712 }
713 assert_eq!(
714 COMPILE_COUNT.with(std::cell::Cell::get),
715 after_create,
716 "evaluate must NOT trigger re-compilation"
717 );
718 }
719
720 #[tokio::test]
721 async fn compilation_happens_only_once_for_predicate() {
722 let before = COMPILE_COUNT.with(std::cell::Cell::get);
723 let lang = JsonPathLanguage::new();
724 let pred = lang.create_predicate("$.items[*]").unwrap();
725 let after_create = COMPILE_COUNT.with(std::cell::Cell::get);
726 assert_eq!(
727 after_create,
728 before + 1,
729 "create_predicate must compile exactly once"
730 );
731
732 for _ in 0..50 {
733 let ex = exchange_with_json(r#"{"items":[1,2,3]}"#).await;
734 assert!(pred.matches(&ex).await.unwrap());
735 }
736 assert_eq!(
737 COMPILE_COUNT.with(std::cell::Cell::get),
738 after_create,
739 "matches must NOT trigger re-compilation"
740 );
741 }
742}