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

camel_processor/data_format/
zip.rs
