camel_processor/data_format/
zip.rs

1use bytes::Bytes;
2use camel_api::body::Body;
3use camel_api::data_format::DataFormat;
4use camel_api::error::CamelError;
5use serde::Deserialize;
6use std::io::Read;
7use std::io::Write;
8use zip::ZipArchive;
9
10const DEFAULT_MAX_DECOMPRESSED_SIZE: u64 = 1_073_741_824;
11/// Default cap on the materialized input size of `marshal` (R3-L1). The eager
12/// marshal collects the whole body into a `Vec<u8>` before compression; this
13/// bounds that allocation.
14const DEFAULT_MAX_INPUT_SIZE: u64 = 64 * 1024 * 1024; // 64 MiB
15const ENTRY_NAME: &str = "payload";
16
17#[derive(Debug, Clone, Deserialize)]
18#[serde(default, deny_unknown_fields)]
19pub struct ZipConfig {
20    pub max_decompressed_size: u64,
21    /// Maximum materialized input size accepted by `marshal` (DoS cap, R3-L1).
22    pub max_input_size: u64,
23    pub compression_level: Option<i32>,
24    pub allow_multi_entry: bool,
25}
26
27impl Default for ZipConfig {
28    fn default() -> Self {
29        Self {
30            max_decompressed_size: DEFAULT_MAX_DECOMPRESSED_SIZE,
31            max_input_size: DEFAULT_MAX_INPUT_SIZE,
32            compression_level: None,
33            allow_multi_entry: false,
34        }
35    }
36}
37
38#[derive(Debug, Clone, Default)]
39pub struct ZipDataFormat {
40    config: ZipConfig,
41}
42
43impl ZipDataFormat {
44    pub fn new(config: ZipConfig) -> Self {
45        Self { config }
46    }
47}
48
49impl DataFormat for ZipDataFormat {
50    fn name(&self) -> &str {
51        "zip"
52    }
53
54    fn marshal(&self, body: Body) -> Result<Body, CamelError> {
55        let content: Vec<u8> = match &body {
56            Body::Text(s) => s.as_bytes().to_vec(),
57            Body::Json(v) => serde_json::to_vec(v).map_err(|e| {
58                CamelError::TypeConversionFailed(format!(
59                    "ZipDataFormat::marshal cannot serialize JSON: {e}"
60                ))
61            })?,
62            Body::Bytes(b) => b.to_vec(),
63            Body::Xml(s) => s.as_bytes().to_vec(),
64            Body::Empty => {
65                return Err(CamelError::TypeConversionFailed(
66                    "ZipDataFormat::marshal requires non-empty body".to_string(),
67                ));
68            }
69            Body::Stream(_) => {
70                return Err(CamelError::TypeConversionFailed(
71                    "cannot marshal Body::Stream — add 'stream_cache' or 'convert_body_to' before this step".to_string(),
72                ));
73            }
74        };
75
76        if content.len() as u64 > self.config.max_input_size {
77            return Err(CamelError::TypeConversionFailed(format!(
78                "ZipDataFormat::marshal input {} bytes exceeds max_input_size {}",
79                content.len(),
80                self.config.max_input_size
81            )));
82        }
83
84        let mut buf = Vec::new();
85        {
86            let mut writer = zip::ZipWriter::new(std::io::Cursor::new(&mut buf));
87            let mut options = zip::write::SimpleFileOptions::default()
88                .compression_method(zip::CompressionMethod::Deflated);
89            if let Some(level) = self.config.compression_level {
90                if !(0..=9).contains(&level) {
91                    return Err(CamelError::TypeConversionFailed(format!(
92                        "ZipDataFormat::marshal compression_level must be 0-9, got {level}"
93                    )));
94                }
95                options = options.compression_level(Some(level as i64));
96            }
97            writer.start_file(ENTRY_NAME, options).map_err(|e| {
98                CamelError::TypeConversionFailed(format!(
99                    "ZipDataFormat::marshal failed to start entry: {e}"
100                ))
101            })?;
102            writer.write_all(&content).map_err(|e| {
103                CamelError::TypeConversionFailed(format!(
104                    "ZipDataFormat::marshal failed to write entry: {e}"
105                ))
106            })?;
107            writer.finish().map_err(|e| {
108                CamelError::TypeConversionFailed(format!(
109                    "ZipDataFormat::marshal failed to finalize archive: {e}"
110                ))
111            })?;
112        }
113
114        Ok(Body::Bytes(Bytes::from(buf)))
115    }
116
117    fn unmarshal(&self, body: Body) -> Result<Body, CamelError> {
118        let raw: Vec<u8> = match &body {
119            Body::Bytes(b) => b.to_vec(),
120            Body::Text(s) => s.as_bytes().to_vec(),
121            Body::Empty => {
122                return Err(CamelError::TypeConversionFailed(
123                    "ZipDataFormat::unmarshal requires non-empty body".to_string(),
124                ));
125            }
126            Body::Stream(_) => {
127                return Err(CamelError::TypeConversionFailed(
128                    "cannot unmarshal Body::Stream — use UnmarshalService which auto-materializes"
129                        .to_string(),
130                ));
131            }
132            Body::Json(_) | Body::Xml(_) => {
133                return Err(CamelError::TypeConversionFailed(
134                    "ZipDataFormat::unmarshal only supports Body::Bytes and Body::Text (ZIP data)"
135                        .to_string(),
136                ));
137            }
138        };
139
140        let reader = std::io::Cursor::new(&raw);
141        let mut archive = ZipArchive::new(reader).map_err(|e| {
142            CamelError::TypeConversionFailed(format!("ZipDataFormat::unmarshal invalid ZIP: {e}"))
143        })?;
144
145        if archive.is_empty() {
146            return Err(CamelError::TypeConversionFailed(
147                "ZipDataFormat::unmarshal ZIP archive has no entries".to_string(),
148            ));
149        }
150
151        if archive.len() > 1 && !self.config.allow_multi_entry {
152            return Err(CamelError::TypeConversionFailed(format!(
153                "ZipDataFormat::unmarshal ZIP has {} entries but allow_multi_entry is false",
154                archive.len()
155            )));
156        }
157
158        if archive.len() > 1 {
159            tracing::warn!(
160                entries = archive.len(),
161                "ZIP archive has multiple entries, extracting first only"
162            );
163        }
164
165        let mut entry = archive.by_index(0).map_err(|e| {
166            CamelError::TypeConversionFailed(format!(
167                "ZipDataFormat::unmarshal failed to read entry: {e}"
168            ))
169        })?;
170
171        let mut decompressed = Vec::new();
172        let limit = self.config.max_decompressed_size.saturating_add(1);
173        let mut limited = std::io::Read::take(&mut entry, limit);
174        limited.read_to_end(&mut decompressed).map_err(|e| {
175            CamelError::TypeConversionFailed(format!(
176                "ZipDataFormat::unmarshal failed to decompress: {e}"
177            ))
178        })?;
179
180        if decompressed.len() as u64 > self.config.max_decompressed_size {
181            return Err(CamelError::TypeConversionFailed(format!(
182                "ZipDataFormat::unmarshal decompressed size exceeds max {}",
183                self.config.max_decompressed_size
184            )));
185        }
186
187        Ok(Body::Bytes(Bytes::from(decompressed)))
188    }
189}
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194    use bytes::Bytes;
195    use serde_json::json;
196    use std::io::Cursor;
197    use std::io::Read;
198    use zip::ZipArchive;
199
200    fn extract_single_entry(zip_bytes: &[u8]) -> Vec<u8> {
201        let reader = Cursor::new(zip_bytes);
202        let mut archive = ZipArchive::new(reader).unwrap();
203        let mut entry = archive.by_index(0).unwrap();
204        let name = entry.name().to_string();
205        assert_eq!(name, "payload");
206        let mut buf = Vec::new();
207        entry.read_to_end(&mut buf).unwrap();
208        buf
209    }
210
211    #[test]
212    fn test_name() {
213        let df = ZipDataFormat::default();
214        assert_eq!(df.name(), "zip");
215    }
216
217    #[test]
218    fn test_marshal_text_to_zip() {
219        let df = ZipDataFormat::default();
220        let body = Body::Text("hello world".to_string());
221        let result = df.marshal(body).unwrap();
222        match result {
223            Body::Bytes(b) => {
224                let decompressed = extract_single_entry(&b);
225                assert_eq!(decompressed, b"hello world");
226            }
227            _ => panic!("expected Body::Bytes"),
228        }
229    }
230
231    #[test]
232    fn test_marshal_json_to_zip() {
233        let df = ZipDataFormat::default();
234        let body = Body::Json(json!({"key": "value"}));
235        let result = df.marshal(body).unwrap();
236        match result {
237            Body::Bytes(b) => {
238                let decompressed = extract_single_entry(&b);
239                let original = serde_json::to_vec(&json!({"key": "value"})).unwrap();
240                assert_eq!(decompressed, original);
241            }
242            _ => panic!("expected Body::Bytes"),
243        }
244    }
245
246    #[test]
247    fn test_marshal_bytes_to_zip() {
248        let df = ZipDataFormat::default();
249        let body = Body::Bytes(Bytes::from_static(b"raw bytes"));
250        let result = df.marshal(body).unwrap();
251        match result {
252            Body::Bytes(b) => {
253                let decompressed = extract_single_entry(&b);
254                assert_eq!(decompressed, b"raw bytes");
255            }
256            _ => panic!("expected Body::Bytes"),
257        }
258    }
259
260    #[test]
261    fn test_marshal_xml_to_zip() {
262        let df = ZipDataFormat::default();
263        let body = Body::Xml("<root><item>1</item></root>".to_string());
264        let result = df.marshal(body).unwrap();
265        match result {
266            Body::Bytes(b) => {
267                let decompressed = extract_single_entry(&b);
268                assert_eq!(decompressed, b"<root><item>1</item></root>");
269            }
270            _ => panic!("expected Body::Bytes"),
271        }
272    }
273
274    #[test]
275    fn test_marshal_empty_error() {
276        let df = ZipDataFormat::default();
277        let result = df.marshal(Body::Empty);
278        assert!(result.is_err());
279    }
280
281    #[test]
282    fn test_marshal_stream_error() {
283        use camel_api::body::{StreamBody, StreamMetadata};
284        use futures::stream;
285        use std::sync::Arc;
286        use tokio::sync::Mutex;
287
288        let stream = stream::iter(vec![Ok(Bytes::from_static(b"data"))]);
289        let body = Body::Stream(StreamBody {
290            stream: Arc::new(Mutex::new(Some(Box::pin(stream)))),
291            metadata: StreamMetadata::default(),
292        });
293        let df = ZipDataFormat::default();
294        let result = df.marshal(body);
295        assert!(result.is_err());
296    }
297
298    fn make_zip(content: &[u8]) -> Vec<u8> {
299        let mut buf = Vec::new();
300        {
301            let mut writer = zip::ZipWriter::new(std::io::Cursor::new(&mut buf));
302            let options = zip::write::SimpleFileOptions::default()
303                .compression_method(zip::CompressionMethod::Deflated);
304            writer.start_file("payload", options).unwrap();
305            writer.write_all(content).unwrap();
306            writer.finish().unwrap();
307        }
308        buf
309    }
310
311    #[test]
312    fn test_unmarshal_zip_bytes() {
313        let df = ZipDataFormat::default();
314        let zip_data = make_zip(b"decompressed content");
315        let body = Body::Bytes(Bytes::from(zip_data));
316        let result = df.unmarshal(body).unwrap();
317        match result {
318            Body::Bytes(b) => assert_eq!(b.as_ref(), b"decompressed content"),
319            _ => panic!("expected Body::Bytes"),
320        }
321    }
322
323    #[test]
324    fn test_unmarshal_zip_text() {
325        let df = ZipDataFormat::default();
326        let content = b"text from text body";
327        let zip_data = make_zip(content);
328        let body = Body::Bytes(Bytes::from(zip_data));
329        let result = df.unmarshal(body).unwrap();
330        match result {
331            Body::Bytes(b) => assert_eq!(b.as_ref(), content),
332            _ => panic!("expected Body::Bytes"),
333        }
334    }
335
336    #[test]
337    fn test_unmarshal_invalid_zip_error() {
338        let df = ZipDataFormat::default();
339        let body = Body::Bytes(Bytes::from_static(b"not a zip file"));
340        let result = df.unmarshal(body);
341        assert!(result.is_err());
342    }
343
344    #[test]
345    fn test_unmarshal_empty_zip_error() {
346        let mut buf = Vec::new();
347        {
348            let writer = zip::ZipWriter::new(std::io::Cursor::new(&mut buf));
349            writer.finish().unwrap();
350        }
351        let df = ZipDataFormat::default();
352        let body = Body::Bytes(Bytes::from(buf));
353        let result = df.unmarshal(body);
354        assert!(result.is_err());
355    }
356
357    #[test]
358    fn test_unmarshal_json_error() {
359        let df = ZipDataFormat::default();
360        let body = Body::Json(json!({"not": "zip"}));
361        let result = df.unmarshal(body);
362        assert!(result.is_err());
363    }
364
365    #[test]
366    fn test_unmarshal_xml_error() {
367        let df = ZipDataFormat::default();
368        let body = Body::Xml("<root/>".to_string());
369        let result = df.unmarshal(body);
370        assert!(result.is_err());
371    }
372
373    #[test]
374    fn test_unmarshal_multi_entry_error() {
375        let mut buf = Vec::new();
376        {
377            let mut writer = zip::ZipWriter::new(std::io::Cursor::new(&mut buf));
378            let options = zip::write::SimpleFileOptions::default();
379            writer.start_file("file1.txt", options).unwrap();
380            writer.write_all(b"one").unwrap();
381            writer.start_file("file2.txt", options).unwrap();
382            writer.write_all(b"two").unwrap();
383            writer.finish().unwrap();
384        }
385        let df = ZipDataFormat::default();
386        let body = Body::Bytes(Bytes::from(buf));
387        let result = df.unmarshal(body);
388        assert!(result.is_err());
389    }
390
391    #[test]
392    fn test_unmarshal_multi_entry_allowed() {
393        let mut buf = Vec::new();
394        {
395            let mut writer = zip::ZipWriter::new(std::io::Cursor::new(&mut buf));
396            let options = zip::write::SimpleFileOptions::default();
397            writer.start_file("file1.txt", options).unwrap();
398            writer.write_all(b"first").unwrap();
399            writer.start_file("file2.txt", options).unwrap();
400            writer.write_all(b"second").unwrap();
401            writer.finish().unwrap();
402        }
403        let config = ZipConfig {
404            allow_multi_entry: true,
405            ..Default::default()
406        };
407        let df = ZipDataFormat::new(config);
408        let body = Body::Bytes(Bytes::from(buf));
409        let result = df.unmarshal(body).unwrap();
410        match result {
411            Body::Bytes(b) => assert_eq!(b.as_ref(), b"first"),
412            _ => panic!("expected Body::Bytes"),
413        }
414    }
415
416    #[test]
417    fn test_roundtrip_text() {
418        let df = ZipDataFormat::default();
419        let original = Body::Text("roundtrip text content".to_string());
420        let compressed = df.marshal(original).unwrap();
421        let decompressed = df.unmarshal(compressed).unwrap();
422        match decompressed {
423            Body::Bytes(b) => assert_eq!(b.as_ref(), b"roundtrip text content"),
424            _ => panic!("expected Body::Bytes"),
425        }
426    }
427
428    #[test]
429    fn test_roundtrip_json() {
430        let df = ZipDataFormat::default();
431        let original = Body::Json(json!({"round": "trip"}));
432        let compressed = df.marshal(original).unwrap();
433        let decompressed = df.unmarshal(compressed).unwrap();
434        match decompressed {
435            Body::Bytes(b) => {
436                let v: serde_json::Value = serde_json::from_slice(&b).unwrap();
437                assert_eq!(v, json!({"round": "trip"}));
438            }
439            _ => panic!("expected Body::Bytes"),
440        }
441    }
442
443    #[test]
444    fn test_roundtrip_bytes() {
445        let df = ZipDataFormat::default();
446        let original = Body::Bytes(Bytes::from_static(b"\x00\x01\x02\xff"));
447        let compressed = df.marshal(original).unwrap();
448        let decompressed = df.unmarshal(compressed).unwrap();
449        match decompressed {
450            Body::Bytes(b) => assert_eq!(b.as_ref(), b"\x00\x01\x02\xff"),
451            _ => panic!("expected Body::Bytes"),
452        }
453    }
454
455    #[test]
456    fn test_max_decompressed_size_exceeded() {
457        let config = ZipConfig {
458            max_decompressed_size: 10,
459            ..Default::default()
460        };
461        let df = ZipDataFormat::new(config);
462        let zip_data = make_zip(b"this content is way longer than 10 bytes");
463        let body = Body::Bytes(Bytes::from(zip_data));
464        let result = df.unmarshal(body);
465        assert!(result.is_err());
466    }
467
468    #[test]
469    fn test_unmarshal_empty_error() {
470        let df = ZipDataFormat::default();
471        let result = df.unmarshal(Body::Empty);
472        assert!(result.is_err());
473    }
474
475    #[test]
476    fn test_unmarshal_stream_error() {
477        use camel_api::body::{StreamBody, StreamMetadata};
478        use futures::stream;
479        use std::sync::Arc;
480        use tokio::sync::Mutex;
481
482        let stream = stream::iter(vec![Ok(Bytes::from_static(b"data"))]);
483        let body = Body::Stream(StreamBody {
484            stream: Arc::new(Mutex::new(Some(Box::pin(stream)))),
485            metadata: StreamMetadata::default(),
486        });
487        let df = ZipDataFormat::default();
488        let result = df.unmarshal(body);
489        assert!(result.is_err());
490    }
491
492    #[test]
493    fn test_marshal_invalid_compression_level() {
494        let config = ZipConfig {
495            compression_level: Some(42),
496            ..Default::default()
497        };
498        let df = ZipDataFormat::new(config);
499        let result = df.marshal(Body::Text("test".to_string()));
500        assert!(result.is_err());
501    }
502
503    #[test]
504    fn test_marshal_input_size_cap() {
505        let config = ZipConfig {
506            max_input_size: 16,
507            ..Default::default()
508        };
509        let df = ZipDataFormat::new(config);
510        let body = Body::Text("x".repeat(64));
511        let result = df.marshal(body);
512        assert!(result.is_err());
513        let msg = format!("{}", result.unwrap_err());
514        assert!(
515            msg.contains("max_input_size"),
516            "error should mention max_input_size: {msg}"
517        );
518    }
519
520    #[test]
521    fn test_marshal_default_cap_accepts_small_input() {
522        let df = ZipDataFormat::default();
523        let body = Body::Text("hello".to_string());
524        let result = df.marshal(body).unwrap();
525        assert!(matches!(result, Body::Bytes(_)));
526    }
527
528    #[test]
529    fn test_builtin_zip_registered() {
530        let df = super::super::builtin_data_format("zip");
531        assert!(df.is_some());
532        assert_eq!(df.unwrap().name(), "zip");
533    }
534
535    #[test]
536    fn test_zip_config_deserialize_from_json() {
537        let json = serde_json::json!({
538            "max_decompressed_size": 2147483648u64,
539            "max_input_size": 134217728u64,
540            "compression_level": 6,
541            "allow_multi_entry": true
542        });
543        let cfg: ZipConfig = serde_json::from_value(json).unwrap();
544        assert_eq!(cfg.max_decompressed_size, 2147483648);
545        assert_eq!(cfg.compression_level, Some(6));
546        assert!(cfg.allow_multi_entry);
547    }
548
549    #[test]
550    fn test_zip_config_deny_unknown_fields() {
551        let json = serde_json::json!({"unknown_key": 42});
552        let result: Result<ZipConfig, _> = serde_json::from_value(json);
553        assert!(result.is_err());
554    }
555}
camel_processor/data_format/zip.rs

camel_processor/data_format/
zip.rs
