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