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

camel_processor/data_format/
zip.rs
