camel_processor/data_format/
tar.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;
7
8const DEFAULT_MAX_DECOMPRESSED_SIZE: u64 = 1_073_741_824;
9/// Default cap on the materialized input size of `marshal` (R3-L1). The eager
10/// marshal collects the whole body into a `Vec<u8>` before archiving; this
11/// bounds that allocation.
12const DEFAULT_MAX_INPUT_SIZE: u64 = 64 * 1024 * 1024; // 64 MiB
13const ENTRY_NAME: &str = "payload";
14
15#[derive(Debug, Clone, Deserialize)]
16#[serde(default, deny_unknown_fields)]
17pub struct TarConfig {
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 allow_multi_entry: bool,
22}
23
24impl Default for TarConfig {
25    fn default() -> Self {
26        Self {
27            max_decompressed_size: DEFAULT_MAX_DECOMPRESSED_SIZE,
28            max_input_size: DEFAULT_MAX_INPUT_SIZE,
29            allow_multi_entry: false,
30        }
31    }
32}
33
34#[derive(Debug, Clone, Default)]
35pub struct TarDataFormat {
36    config: TarConfig,
37}
38
39impl TarDataFormat {
40    pub fn new(config: TarConfig) -> Self {
41        Self { config }
42    }
43}
44
45impl DataFormat for TarDataFormat {
46    fn name(&self) -> &str {
47        "tar"
48    }
49
50    fn marshal(&self, body: Body) -> Result<Body, CamelError> {
51        let content =
52            super::materialize_marshal_input("TarDataFormat", &body, self.config.max_input_size)?;
53
54        let mut buf = Vec::new();
55        {
56            let mut builder = tar::Builder::new(&mut buf);
57            let mut header = tar::Header::new_gnu();
58            header.set_entry_type(tar::EntryType::Regular);
59            header.set_size(content.len() as u64);
60            header.set_mode(0o644);
61            header.set_cksum();
62            builder
63                .append_data(&mut header, ENTRY_NAME, content.as_slice())
64                .map_err(|e| {
65                    CamelError::TypeConversionFailed(format!(
66                        "TarDataFormat::marshal failed to write entry: {e}"
67                    ))
68                })?;
69            builder.finish().map_err(|e| {
70                CamelError::TypeConversionFailed(format!(
71                    "TarDataFormat::marshal failed to finalize archive: {e}"
72                ))
73            })?;
74        }
75
76        Ok(Body::Bytes(Bytes::from(buf)))
77    }
78
79    fn unmarshal(&self, body: Body) -> Result<Body, CamelError> {
80        let raw = super::raw_unmarshal_body("TarDataFormat", "TAR data", &body)?;
81
82        // TAR has no central directory: walk entries in stream order. Names
83        // are never used for I/O and non-regular entries are skipped without
84        // materializing their payload, so archive paths cannot escape
85        // (path-confinement precedent rc-0ks57; v1 performs no disk I/O).
86        let mut archive = tar::Archive::new(std::io::Cursor::new(&raw));
87        let mut first_regular: Option<Vec<u8>> = None;
88        let mut regular_count: usize = 0;
89        {
90            let entries = archive.entries().map_err(|e| {
91                CamelError::TypeConversionFailed(format!(
92                    "TarDataFormat::unmarshal invalid TAR: {e}"
93                ))
94            })?;
95            for entry in entries {
96                let entry = entry.map_err(|e| {
97                    CamelError::TypeConversionFailed(format!(
98                        "TarDataFormat::unmarshal invalid TAR entry: {e}"
99                    ))
100                })?;
101                if entry.header().entry_type() != tar::EntryType::Regular {
102                    continue;
103                }
104                regular_count += 1;
105                if first_regular.is_none() {
106                    // Read at most cap + 1 bytes so oversize payloads are
107                    // detected without unbounded materialization.
108                    let limit = self.config.max_decompressed_size.saturating_add(1);
109                    let mut limited = entry.take(limit);
110                    let mut data = Vec::new();
111                    limited.read_to_end(&mut data).map_err(|e| {
112                        CamelError::TypeConversionFailed(format!(
113                            "TarDataFormat::unmarshal failed to read entry: {e}"
114                        ))
115                    })?;
116                    first_regular = Some(data);
117                }
118            }
119        }
120
121        let payload = first_regular.ok_or_else(|| {
122            CamelError::TypeConversionFailed(
123                "TarDataFormat::unmarshal TAR archive has no regular file".to_string(),
124            )
125        })?;
126
127        if regular_count > 1 && !self.config.allow_multi_entry {
128            return Err(CamelError::TypeConversionFailed(format!(
129                "TarDataFormat::unmarshal TAR has {regular_count} regular files but allow_multi_entry is false"
130            )));
131        }
132
133        if regular_count > 1 {
134            tracing::warn!(
135                regular_files = regular_count,
136                "TAR archive has multiple regular files, returning first only"
137            );
138        }
139
140        if payload.len() as u64 > self.config.max_decompressed_size {
141            return Err(CamelError::TypeConversionFailed(format!(
142                "TarDataFormat::unmarshal regular file size exceeds max_decompressed_size {}",
143                self.config.max_decompressed_size
144            )));
145        }
146
147        Ok(Body::Bytes(Bytes::from(payload)))
148    }
149}
150
151#[cfg(test)]
152mod tests {
153    use super::*;
154    use bytes::Bytes;
155
156    use super::super::test_util::{
157        CHARACTER_DEVICE, DIRECTORY, HARDLINK, REGULAR, SYMLINK, assert_bytes, capture_warns,
158        make_tar, make_tar_payload, stream_body_pair,
159    };
160
161    #[test]
162    fn test_name() {
163        let df = TarDataFormat::default();
164        assert_eq!(df.name(), "tar");
165    }
166
167    #[test]
168    fn test_tar_config_deserialize_from_json() {
169        let json = serde_json::json!({
170            "max_decompressed_size": 2147483648u64,
171            "max_input_size": 134217728u64,
172            "allow_multi_entry": true
173        });
174        let cfg: TarConfig = serde_json::from_value(json).unwrap();
175        assert_eq!(cfg.max_decompressed_size, 2147483648);
176        assert_eq!(cfg.max_input_size, 134217728);
177        assert!(cfg.allow_multi_entry);
178    }
179
180    #[test]
181    fn test_tar_config_deny_unknown_fields() {
182        let json = serde_json::json!({"unknown_key": 42});
183        let result: Result<TarConfig, _> = serde_json::from_value(json);
184        assert!(result.is_err());
185    }
186
187    #[test]
188    fn tar_round_trip_bytes() {
189        let df = TarDataFormat::default();
190        let original = Body::Bytes(Bytes::from_static(b"\x00\x01\x02\xffbinary payload"));
191        let archived = df.marshal(original.clone()).unwrap();
192        let archived_bytes = match &archived {
193            Body::Bytes(b) => b.clone(),
194            _ => panic!("expected Body::Bytes"),
195        };
196        let mut names = Vec::new();
197        for entry in tar::Archive::new(std::io::Cursor::new(&archived_bytes[..]))
198            .entries()
199            .unwrap()
200        {
201            names.push(entry.unwrap().path().unwrap().to_path_buf());
202        }
203        assert_eq!(names.len(), 1, "marshal must write exactly one entry");
204        assert_eq!(names[0], std::path::Path::new(ENTRY_NAME));
205        let restored = df.unmarshal(archived).unwrap();
206        assert_eq!(restored, original);
207    }
208
209    #[test]
210    fn tar_regular_entry_policy() {
211        let df_strict = TarDataFormat::default();
212        let df_multi = TarDataFormat::new(TarConfig {
213            allow_multi_entry: true,
214            ..Default::default()
215        });
216
217        // Zero regular files (only directory + symlink) errors under both
218        // configurations.
219        let no_regular = make_tar(&[("only-dir/", DIRECTORY, b""), ("only-link", SYMLINK, b"")]);
220        for df in [&df_strict, &df_multi] {
221            let result = df.unmarshal(Body::Bytes(Bytes::from(no_regular.clone())));
222            match result {
223                Err(CamelError::TypeConversionFailed(_)) => {}
224                _ => panic!("archive without regular files must be rejected"),
225            }
226        }
227
228        // Exactly one regular file (plus non-regular entries) succeeds under
229        // both configurations.
230        let one_regular = make_tar(&[
231            ("dir/", DIRECTORY, b""),
232            ("payload.txt", REGULAR, b"single payload"),
233            ("link", HARDLINK, b""),
234        ]);
235        for df in [&df_strict, &df_multi] {
236            let restored = df
237                .unmarshal(Body::Bytes(Bytes::from(one_regular.clone())))
238                .unwrap();
239            assert_bytes(restored, b"single payload");
240        }
241
242        // Two regular files: strict config errors mentioning the policy;
243        // allow_multi_entry returns the first regular file and warns.
244        let two_regular = make_tar(&[
245            ("first.txt", REGULAR, b"first"),
246            ("dir/", DIRECTORY, b""),
247            ("second.txt", REGULAR, b"second"),
248        ]);
249        let err = df_strict
250            .unmarshal(Body::Bytes(Bytes::from(two_regular)))
251            .unwrap_err();
252        let msg = format!("{err}");
253        assert!(
254            msg.contains("allow_multi_entry"),
255            "error should mention allow_multi_entry: {msg}"
256        );
257
258        let two_regular = make_tar(&[
259            ("first.txt", REGULAR, b"first"),
260            ("second.txt", REGULAR, b"second"),
261        ]);
262        let (result, warnings) =
263            capture_warns(|| df_multi.unmarshal(Body::Bytes(Bytes::from(two_regular.clone()))));
264        assert_bytes(result.unwrap(), b"first");
265        assert!(
266            warnings.iter().any(|w| w.contains("multiple regular")),
267            "multi-entry path should warn, captured: {warnings:?}"
268        );
269    }
270
271    #[test]
272    fn tar_non_regular_entries_and_malicious_paths_are_ignored() {
273        let tar_bytes = make_tar(&[
274            ("../escape-dir/", DIRECTORY, b""),
275            ("/etc/passwd", SYMLINK, b""),
276            ("../escape-hardlink", HARDLINK, b""),
277            ("/dev/tty", CHARACTER_DEVICE, b""),
278            ("payload.txt", REGULAR, b"safe payload bytes"),
279        ]);
280        let df = TarDataFormat::default();
281        // The archive is scanned in memory only: no extraction, no path-based
282        // I/O, so the malicious names above are never resolved. Only the
283        // regular file is selected.
284        let restored = df.unmarshal(Body::Bytes(Bytes::from(tar_bytes))).unwrap();
285        assert_bytes(restored, b"safe payload bytes");
286    }
287
288    #[test]
289    fn malformed_tar_input_rejected() {
290        let df = TarDataFormat::default();
291
292        // Garbage bytes fail header validation (checksum/octal parse).
293        let garbage = vec![b'G'; 1024];
294        let result = df.unmarshal(Body::Bytes(Bytes::from(garbage)));
295        match result {
296            Err(CamelError::TypeConversionFailed(_)) => {}
297            _ => panic!("garbage input must yield TypeConversionFailed"),
298        }
299
300        // Truncated header block fails before any entry is returned.
301        let full = make_tar_payload(b"hello world");
302        let truncated = full[..400].to_vec();
303        let result = df.unmarshal(Body::Bytes(Bytes::from(truncated)));
304        match result {
305            Err(CamelError::TypeConversionFailed(_)) => {}
306            _ => panic!("truncated header must yield TypeConversionFailed"),
307        }
308    }
309
310    #[test]
311    fn tar_materialized_empty_and_stream_bodies() {
312        let df = TarDataFormat::default();
313
314        // Zero-length materialized payloads follow TAR semantics: marshal
315        // yields a valid archive with one empty regular entry; unmarshal
316        // returns the empty payload.
317        for body in [Body::Bytes(Bytes::new()), Body::Text(String::new())] {
318            let archived = df.marshal(body).unwrap();
319            let restored = df.unmarshal(archived).unwrap();
320            assert_bytes(restored, b"");
321        }
322
323        // Body::Empty fails marshal.
324        assert!(df.marshal(Body::Empty).is_err());
325
326        // Body::Stream fails marshal and unmarshal without consuming the
327        // stream.
328        let (body, slot) = stream_body_pair();
329        assert!(df.marshal(body).is_err());
330        assert!(
331            slot.blocking_lock().is_some(),
332            "marshal must not consume the stream"
333        );
334
335        let (body, slot) = stream_body_pair();
336        assert!(df.unmarshal(body).is_err());
337        assert!(
338            slot.blocking_lock().is_some(),
339            "unmarshal must not consume the stream"
340        );
341    }
342
343    #[test]
344    fn test_marshal_input_size_cap() {
345        let config = TarConfig {
346            max_input_size: 16,
347            ..Default::default()
348        };
349        let df = TarDataFormat::new(config);
350        let result = df.marshal(Body::Text("x".repeat(64)));
351        assert!(result.is_err());
352        let msg = format!("{}", result.unwrap_err());
353        assert!(
354            msg.contains("max_input_size"),
355            "error should mention max_input_size: {msg}"
356        );
357    }
358
359    #[test]
360    fn test_max_decompressed_size_exceeded() {
361        let config = TarConfig {
362            max_decompressed_size: 10,
363            ..Default::default()
364        };
365        let df = TarDataFormat::new(config);
366        let tar_data = make_tar_payload(b"this content is way longer than 10 bytes");
367        let result = df.unmarshal(Body::Bytes(Bytes::from(tar_data)));
368        assert!(result.is_err());
369    }
370}
camel_processor/data_format/tar.rs

camel_processor/data_format/
tar.rs
