camel_component_exec/
producer.rs

1//! producer.rs — ExecProducer: Service<Exchange> running the enforcement flow.
2//!
3//! Outcomes: pre/during-spawn failures => Err. Post-spawn conditions with output
4//! (timeout, exit-code mismatch) => Ok with result + headers (route branches via choice).
5
6use std::collections::HashMap;
7use std::sync::Arc;
8use std::time::{Duration, Instant};
9
10use base64::Engine;
11use bytes::Bytes;
12use camel_api::{Body, CamelError, Exchange};
13use camel_component_api::RuntimeObservability;
14use tokio::sync::Semaphore;
15use tower::Service;
16
17use crate::audit::{ExecAuditEvent, emit};
18use crate::config::{ExecGlobalConfig, ExecProfile};
19use crate::error::ExecError;
20use crate::headers::*;
21use crate::policy;
22use crate::process::{self, RawResult};
23
24pub struct ExecProducer {
25    pub(crate) profile: Arc<ExecProfile>,
26    pub(crate) global: Arc<ExecGlobalConfig>,
27    pub(crate) route_id: String,
28    pub(crate) host_env: Arc<HashMap<String, String>>,
29    pub(crate) semaphore: Arc<Semaphore>,
30    pub(crate) rt: Option<Arc<dyn RuntimeObservability>>,
31}
32
33impl Clone for ExecProducer {
34    fn clone(&self) -> Self {
35        Self {
36            profile: Arc::clone(&self.profile),
37            global: Arc::clone(&self.global),
38            route_id: self.route_id.clone(),
39            host_env: Arc::clone(&self.host_env),
40            semaphore: Arc::clone(&self.semaphore),
41            rt: self.rt.clone(),
42        }
43    }
44}
45
46impl Service<Exchange> for ExecProducer {
47    type Response = Exchange;
48    type Error = CamelError;
49    type Future = std::pin::Pin<
50        Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
51    >;
52
53    fn poll_ready(
54        &mut self,
55        _cx: &mut std::task::Context<'_>,
56    ) -> std::task::Poll<Result<(), Self::Error>> {
57        std::task::Poll::Ready(Ok(()))
58    }
59
60    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
61        let this = self.clone();
62        Box::pin(async move { this.run(&mut exchange).await.map(|_| exchange) })
63    }
64}
65
66impl ExecProducer {
67    /// Create a new ExecProducer. Public for integration tests.
68    pub fn new(
69        profile: Arc<ExecProfile>,
70        global: Arc<ExecGlobalConfig>,
71        route_id: String,
72        host_env: HashMap<String, String>,
73        semaphore: Arc<Semaphore>,
74        rt: Option<Arc<dyn RuntimeObservability>>,
75    ) -> Self {
76        Self {
77            profile,
78            global,
79            route_id,
80            host_env: Arc::new(host_env),
81            semaphore,
82            rt,
83        }
84    }
85
86    async fn run(&self, ex: &mut Exchange) -> Result<(), CamelError> {
87        let start = Instant::now();
88        // 1. args
89        let args = self.resolve_args(ex)?;
90        // 2. arg-policy
91        if let Err(e) = policy::eval_args(&args, &self.profile.args, &self.profile.deny_flags) {
92            self.metric_denied("arg_policy");
93            return Err(pack(&self.route_id, e));
94        }
95        // 3. shell reject
96        let exe = self
97            .profile
98            .canonical_executable
99            .clone()
100            .unwrap_or_default();
101        let exe_str = exe.to_string_lossy().to_string();
102        if !self.profile.allow_shell && policy::is_known_shell(&exe_str) {
103            self.metric_denied("shell");
104            return Err(pack(
105                &self.route_id,
106                ExecError::ShellRejected {
107                    executable: exe_str.clone(),
108                },
109            ));
110        }
111        // 4. env build (deny_env applied last, always wins)
112        let env = policy::build_env(&self.profile.env, &self.global.deny_env, &self.host_env);
113        // 5. cwd — use the STARTUP-PINNED canonical root (I-6: no "." fallback)
114        let cwd = match &self.global.canonical_workspace_root {
115            Some(root) => crate::config::confine(root, &self.profile.working_dir).map_err(|e| {
116                self.metric_denied("workdir");
117                pack(&self.route_id, ExecError::InvalidWorkDir { path: e })
118            })?,
119            None => {
120                return Err(pack(
121                    &self.route_id,
122                    ExecError::InvalidWorkDir {
123                        path: "workspace root not pinned at startup".into(),
124                    },
125                ));
126            }
127        };
128        // 6. stdin (capped)
129        let stdin =
130            extract_stdin(ex, self.profile.stdin_max_bytes).map_err(|e| pack(&self.route_id, e))?;
131        // 7. concurrency permit — acquired BEFORE spawn (held through spawn→collect)
132        let _permit = self.semaphore.acquire().await.map_err(|_| {
133            CamelError::ProcessorErrorWithSource(
134                "exec semaphore closed".into(),
135                Arc::new(ExecError::Spawn(std::io::Error::other("semaphore"))),
136            )
137        })?;
138        // 8. spawn + timeout + drain (child held outside the timeout region — C-1)
139        let raw = self
140            .spawn_and_collect(&exe, &args, &env, &cwd, stdin)
141            .await
142            .map_err(|e| pack(&self.route_id, e))?;
143        // 9. populate exchange (non-error outcome even on timeout / non-zero exit)
144        let exit_code = raw.exit_code;
145        let accepted = exit_code
146            .map(|c| self.profile.accepted_exit_codes.contains(&c))
147            .unwrap_or(false);
148        let body_out = ExecResult {
149            exit_code,
150            stdout: b64(&raw.stdout),
151            stderr: b64(&raw.stderr),
152            stdout_truncated: raw.stdout_truncated,
153            stderr_truncated: raw.stderr_truncated,
154            timed_out: raw.timed_out,
155            profile: self.profile.name.clone(),
156            duration_ms: start.elapsed().as_millis() as u64,
157        };
158        ex.input.body = Body::Json(serde_json::to_value(&body_out).map_err(|e| {
159            pack(
160                &self.route_id,
161                ExecError::InvalidArgs(format!("serialize result: {e}")),
162            )
163        })?);
164        let h = &mut ex.input.headers;
165        h.insert(
166            CAMEL_EXEC_PROFILE.to_string(),
167            serde_json::Value::String(self.profile.name.clone()),
168        );
169        // Always emit CamelExecExitAccepted (false when no exit code, e.g. timeout).
170        if let Some(c) = exit_code {
171            h.insert(CAMEL_EXEC_EXIT_CODE.to_string(), serde_json::Value::from(c));
172            h.insert(
173                CAMEL_EXEC_EXIT_ACCEPTED.to_string(),
174                serde_json::Value::from(accepted),
175            );
176        } else {
177            h.insert(
178                CAMEL_EXEC_EXIT_ACCEPTED.to_string(),
179                serde_json::Value::from(false),
180            );
181        }
182        h.insert(
183            CAMEL_EXEC_TIMED_OUT.to_string(),
184            serde_json::Value::from(raw.timed_out),
185        );
186        h.insert(
187            CAMEL_EXEC_STDOUT_TRUNCATED.to_string(),
188            serde_json::Value::from(raw.stdout_truncated),
189        );
190        h.insert(
191            CAMEL_EXEC_STDERR_TRUNCATED.to_string(),
192            serde_json::Value::from(raw.stderr_truncated),
193        );
194        // Header CAMEL_EXEC_STDERR carries lossy-UTF8 for route predicates
195        // (log:, choice()); body ExecResult.stderr carries byte-exact base64.
196        // Dual repr is intentional.
197        h.insert(
198            CAMEL_EXEC_STDERR.to_string(),
199            serde_json::Value::String(String::from_utf8_lossy(&raw.stderr).into_owned()),
200        );
201        // 10. metrics + audit
202        self.metric_outcome(
203            exit_code,
204            raw.timed_out,
205            raw.stdout_truncated,
206            start.elapsed(),
207        );
208        emit(&ExecAuditEvent {
209            route_id: &self.route_id,
210            profile: &self.profile.name,
211            canonical_executable: &exe_str,
212            args: &args,
213            env_keys: env.keys().map(|k| k.as_str()).collect::<Vec<_>>(),
214            cwd: &cwd.to_string_lossy(),
215            exit_code,
216            timed_out: raw.timed_out,
217            stdout_truncated: raw.stdout_truncated,
218            stderr_truncated: raw.stderr_truncated,
219            duration: start.elapsed(),
220        });
221        Ok(())
222    }
223
224    async fn spawn_and_collect(
225        &self,
226        exe: &std::path::Path,
227        args: &[String],
228        env: &HashMap<String, String>,
229        cwd: &std::path::Path,
230        stdin: Bytes,
231    ) -> Result<RawResult, ExecError> {
232        // C-1: spawn OUTSIDE the timeout so the Child handle survives a timeout
233        // and kill_tree can fire. kill_on_drop(true) is a belt-and-suspenders guard.
234        let mut child = process::spawn(exe, args, env, cwd)?;
235        let stdin_handle = child.stdin.take();
236        let out = child.stdout.take();
237        let err = child.stderr.take();
238        // drain + stdin tasks run concurrently; they complete when pipes close.
239        let out_task = tokio::spawn(drain_join(out, self.profile.stdout_max_bytes));
240        let err_task = tokio::spawn(drain_join(err, self.profile.stderr_max_bytes));
241        let stdin_task = tokio::spawn(async move {
242            if let Some(mut sin) = stdin_handle {
243                use tokio::io::AsyncWriteExt;
244                let _ = sin.write_all(&stdin).await;
245                let _ = sin.shutdown().await;
246            }
247        });
248
249        let timeout = Duration::from_secs(self.profile.timeout_secs);
250        // select! keeps `child` accessible after either branch (the losing branch's
251        // future is dropped, releasing the &mut borrow). Timeout bounds stdin-write
252        // + wait together (a child that never reads stdin no longer hangs unbounded).
253        let (exit_code, timed_out) = tokio::select! {
254            biased;
255            status = child.wait() => match status {
256                Ok(s) => (s.code(), false),
257                Err(e) => return Err(ExecError::Spawn(e)),
258            },
259            _ = tokio::time::sleep(timeout) => {
260                process::kill_tree(&mut child);
261                let _ = child.wait().await; // reap the killed tree
262                (None, true)
263            }
264        };
265        // After success or timeout, collect whatever the drain tasks produced.
266        // (On timeout the pipes close after kill → tasks finish quickly.)
267        let _ = tokio::time::timeout(Duration::from_secs(2), stdin_task).await; // best-effort
268        let (stdout, stdout_tr) = out_task.await.unwrap_or((Bytes::new(), false));
269        let (stderr, stderr_tr) = err_task.await.unwrap_or((Bytes::new(), false));
270        Ok(RawResult {
271            stdout,
272            stderr,
273            stdout_truncated: stdout_tr,
274            stderr_truncated: stderr_tr,
275            exit_code,
276            timed_out,
277        })
278    }
279
280    fn resolve_args(&self, ex: &Exchange) -> Result<Vec<String>, CamelError> {
281        if let Some(v) = ex.input.headers.get(CAMEL_EXEC_ARGS) {
282            serde_json::from_value::<Vec<String>>(v.clone())
283                .map_err(|e| pack(&self.route_id, ExecError::InvalidArgs(e.to_string())))
284        } else {
285            Ok(Vec::new())
286        }
287    }
288
289    fn metric_denied(&self, reason: &'static str) {
290        if let Some(rt) = &self.rt {
291            // allow-open-label rc-xl5k rc-gm6s (route; reason bounded by caller literals)
292            rt.metrics().record_counter(
293                "exec_policy_denials_total",
294                1.0,
295                &[("reason", reason), ("route", &self.route_id)],
296            );
297        }
298    }
299
300    fn metric_outcome(
301        &self,
302        exit_code: Option<i32>,
303        timed_out: bool,
304        stdout_truncated: bool,
305        dur: Duration,
306    ) {
307        let Some(rt) = &self.rt else { return };
308        // allow-open-label rc-xl5k (route label: user-defined route id)
309        rt.metrics().record_histogram(
310            "exec_duration_secs",
311            dur.as_secs_f64(),
312            &[("route", &self.route_id)],
313        );
314        if let Some(c) = exit_code {
315            let code_str = c.to_string();
316            // allow-open-label rc-xl5k rc-gm6s (route; exit code bounded 0-255)
317            rt.metrics().record_counter(
318                "exec_exit_code",
319                1.0,
320                &[("code", &code_str), ("route", &self.route_id)],
321            );
322        }
323        if timed_out {
324            // allow-open-label rc-xl5k (route label: user-defined route id)
325            rt.metrics()
326                .record_counter("exec_timeouts_total", 1.0, &[("route", &self.route_id)]);
327        }
328        if stdout_truncated {
329            // allow-open-label rc-xl5k (route label: user-defined route id)
330            rt.metrics().record_counter(
331                "exec_stdout_truncated_total",
332                1.0,
333                &[("route", &self.route_id)],
334            );
335        }
336    }
337}
338
339/// Drain an Option<ChildStdout/Stderr> to (Bytes, truncated). Spawned as a task.
340async fn drain_join<R: tokio::io::AsyncRead + Unpin + Send + 'static>(
341    r: Option<R>,
342    cap: usize,
343) -> (Bytes, bool) {
344    match r {
345        Some(reader) => process::drain_with_cap(reader, cap).await,
346        None => (Bytes::new(), false),
347    }
348}
349
350/// Materialized result placed in the body. stdout/stderr are base64-encoded
351/// Strings (C-2: Bytes is not Serialize without a serde feature, and raw byte
352/// arrays would be a pathological JSON). `#[non_exhaustive]`.
353#[derive(Debug, serde::Serialize, serde::Deserialize)]
354#[non_exhaustive]
355pub struct ExecResult {
356    pub exit_code: Option<i32>,
357    pub stdout: String, // base64
358    pub stderr: String, // base64
359    pub stdout_truncated: bool,
360    pub stderr_truncated: bool,
361    pub timed_out: bool,
362    pub profile: String,
363    pub duration_ms: u64,
364}
365
366fn b64(b: &[u8]) -> String {
367    base64::engine::general_purpose::STANDARD.encode(b)
368}
369
370fn extract_stdin(ex: &Exchange, cap: usize) -> Result<Bytes, ExecError> {
371    let b = match &ex.input.body {
372        Body::Text(s) => Bytes::copy_from_slice(s.as_bytes()),
373        Body::Bytes(b) => Bytes::copy_from_slice(b),
374        Body::Json(v) => Bytes::from(serde_json::to_vec(v).unwrap_or_default()),
375        _ => Bytes::new(),
376    };
377    if b.len() > cap {
378        return Err(ExecError::StdinTooLarge {
379            size: b.len(),
380            max: cap,
381        });
382    }
383    Ok(b)
384}
385
386fn pack(route_id: &str, e: ExecError) -> CamelError {
387    CamelError::ProcessorErrorWithSource(format!("exec producer ({route_id}): {e}"), Arc::new(e))
388}
389
390#[cfg(test)]
391mod tests {
392    use super::*;
393    use crate::config::{ArgPolicy, ExecGlobalConfig, ExecProfile};
394    use camel_api::{Body, Exchange};
395    use std::collections::HashMap;
396    use std::path::PathBuf;
397    use std::sync::Arc;
398    use tokio::sync::Semaphore;
399
400    /// Minimal PATH lookup (mirrors config::which, but does NOT canonicalize).
401    /// Tests need the symlink name (e.g., "echo") preserved as argv[0] for multicall binaries.
402    fn which(name: &str) -> PathBuf {
403        let path = std::env::var_os("PATH").expect("PATH must be set");
404        for dir in std::env::split_paths(&path) {
405            let cand = dir.join(name);
406            if cand.is_file() {
407                return cand; // Do NOT canonicalize — preserve symlink name for argv[0]
408            }
409        }
410        panic!("{name} not found on PATH");
411    }
412
413    fn make_global() -> Arc<ExecGlobalConfig> {
414        let root = std::env::temp_dir().canonicalize().unwrap();
415        Arc::new(ExecGlobalConfig {
416            canonical_workspace_root: Some(root),
417            ..ExecGlobalConfig::default()
418        })
419    }
420
421    fn make_profile(
422        exe: &str,
423        args_policy: ArgPolicy,
424        accepted: Vec<i32>,
425        timeout: u64,
426    ) -> ExecProfile {
427        // Provide minimal env: set PATH explicitly so binaries can be found.
428        let mut env = crate::config::EnvPolicy::default();
429        if let Ok(path) = std::env::var("PATH") {
430            env.set.insert("PATH".into(), path);
431        }
432
433        ExecProfile {
434            name: "test".into(),
435            executable: exe.into(),
436            args: args_policy,
437            deny_flags: Default::default(),
438            allow_shell: true, // allow sh for tests
439            env,
440            working_dir: ".".into(),
441            timeout_secs: timeout,
442            accepted_exit_codes: accepted,
443            concurrency: 1,
444            stdin_max_bytes: 1 << 20,
445            stdout_max_bytes: 10 << 20,
446            stderr_max_bytes: 1 << 20,
447            sandbox: Default::default(),
448            output_mode: Default::default(),
449            canonical_executable: Some(which(exe)),
450        }
451    }
452
453    fn make_producer(profile: ExecProfile) -> ExecProducer {
454        ExecProducer {
455            profile: Arc::new(profile),
456            global: make_global(),
457            route_id: "test".into(),
458            host_env: Arc::new(HashMap::new()),
459            semaphore: Arc::new(Semaphore::new(1)),
460            rt: None,
461        }
462    }
463
464    /// Helper: set CamelExecArgs header on the exchange.
465    fn set_args(ex: &mut Exchange, args: Vec<&str>) {
466        let v: Vec<String> = args.into_iter().map(String::from).collect();
467        ex.input.headers.insert(
468            crate::headers::CAMEL_EXEC_ARGS.to_string(),
469            serde_json::to_value(v).unwrap(),
470        );
471    }
472
473    /// Helper: deserialize the ExecResult JSON from the exchange body.
474    fn result_from_body(ex: &Exchange) -> ExecResult {
475        let v = match &ex.input.body {
476            Body::Json(v) => v.clone(),
477            other => panic!("expected Body::Json, got {other:?}"),
478        };
479        serde_json::from_value(v).expect("deserialize ExecResult")
480    }
481
482    #[tokio::test]
483    async fn echo_happy_path() {
484        let profile = make_profile("echo", ArgPolicy::Any, vec![0], 30);
485        let producer = make_producer(profile);
486        let mut ex = Exchange::default();
487        set_args(&mut ex, vec!["hello"]);
488
489        producer.run(&mut ex).await.expect("echo must succeed");
490
491        let result = result_from_body(&ex);
492        assert_eq!(result.exit_code, Some(0));
493        assert!(!result.timed_out);
494        // stdout is base64-encoded "hello\n"
495        let stdout_bytes = base64::engine::general_purpose::STANDARD
496            .decode(&result.stdout)
497            .unwrap();
498        assert_eq!(stdout_bytes, b"hello\n");
499        // Header assertions
500        assert_eq!(
501            ex.input
502                .headers
503                .get(crate::headers::CAMEL_EXEC_EXIT_ACCEPTED),
504            Some(&serde_json::Value::Bool(true))
505        );
506        assert_eq!(
507            ex.input.headers.get(crate::headers::CAMEL_EXEC_TIMED_OUT),
508            Some(&serde_json::Value::Bool(false))
509        );
510    }
511
512    #[tokio::test]
513    async fn timeout_is_non_error_outcome() {
514        let profile = make_profile("sleep", ArgPolicy::Any, vec![0], 1);
515        let producer = make_producer(profile);
516        let mut ex = Exchange::default();
517        set_args(&mut ex, vec!["30"]);
518
519        // KEY ASSERTION: timeout returns Ok(()), NOT Err.
520        let res = producer.run(&mut ex).await;
521        assert!(
522            res.is_ok(),
523            "timeout must be non-error outcome, got {res:?}"
524        );
525
526        let result = result_from_body(&ex);
527        assert_eq!(result.exit_code, None, "killed process has no exit code");
528        assert!(result.timed_out, "must be marked as timed out");
529        // CamelExecExitAccepted must be false when no exit code
530        assert_eq!(
531            ex.input
532                .headers
533                .get(crate::headers::CAMEL_EXEC_EXIT_ACCEPTED),
534            Some(&serde_json::Value::Bool(false))
535        );
536        assert_eq!(
537            ex.input.headers.get(crate::headers::CAMEL_EXEC_TIMED_OUT),
538            Some(&serde_json::Value::Bool(true))
539        );
540    }
541
542    #[tokio::test]
543    async fn non_zero_exit_is_non_error_outcome() {
544        // Case 1: accepted_exit_codes=[0], actual exit=2 → accepted=false
545        let profile = make_profile("sh", ArgPolicy::Any, vec![0], 30);
546        let producer = make_producer(profile);
547        let mut ex = Exchange::default();
548        set_args(&mut ex, vec!["-c", "exit 2"]);
549
550        let res = producer.run(&mut ex).await;
551        assert!(res.is_ok(), "non-zero exit must be non-error outcome");
552
553        let result = result_from_body(&ex);
554        assert_eq!(result.exit_code, Some(2));
555        assert!(!result.timed_out);
556        assert_eq!(
557            ex.input
558                .headers
559                .get(crate::headers::CAMEL_EXEC_EXIT_ACCEPTED),
560            Some(&serde_json::Value::Bool(false)),
561            "exit 2 not in accepted_exit_codes=[0]"
562        );
563
564        // Case 2: accepted_exit_codes=[2], actual exit=2 → accepted=true
565        let profile2 = make_profile("sh", ArgPolicy::Any, vec![2], 30);
566        let producer2 = make_producer(profile2);
567        let mut ex2 = Exchange::default();
568        set_args(&mut ex2, vec!["-c", "exit 2"]);
569
570        producer2.run(&mut ex2).await.expect("must succeed");
571        assert_eq!(
572            ex2.input
573                .headers
574                .get(crate::headers::CAMEL_EXEC_EXIT_ACCEPTED),
575            Some(&serde_json::Value::Bool(true)),
576            "exit 2 in accepted_exit_codes=[2]"
577        );
578    }
579
580    #[test]
581    fn clone_shares_host_env() {
582        let producer = make_producer(make_profile("echo", ArgPolicy::Any, vec![0], 30));
583        let cloned = producer.clone();
584        assert!(
585            Arc::ptr_eq(&producer.host_env, &cloned.host_env),
586            "clone must share host_env Arc, not deep-copy"
587        );
588    }
589
590    #[tokio::test]
591    async fn drain_with_cap_stores_prefix_on_overflow() {
592        // &[u8] delivers all 100 bytes in a single read (tmp buf is 8192),
593        // so this exercises the n > space branch (store space, flip truncated).
594        let data: &[u8] = &[b'x'; 100];
595        let (buf, truncated) = process::drain_with_cap(data, 50).await;
596        assert_eq!(buf.len(), 50, "buf must contain exactly cap bytes");
597        assert!(truncated);
598    }
599
600    #[tokio::test]
601    async fn drain_with_cap_multi_read_boundary() {
602        // Chain two slices to force two reads: 30 bytes then 30 bytes.
603        // First read fits (buf=30), second read overflows (space=20 → store 20, flip).
604        use tokio::io::AsyncReadExt;
605        let r1: &[u8] = &[b'x'; 30];
606        let r2: &[u8] = &[b'y'; 30];
607        let chained = r1.chain(r2);
608        let (buf, truncated) = process::drain_with_cap(chained, 50).await;
609        assert_eq!(buf.len(), 50);
610        assert!(truncated);
611    }
612
613    #[tokio::test]
614    async fn drain_with_cap_exact_fit_not_truncated() {
615        let data: &[u8] = &[b'x'; 50];
616        let (buf, truncated) = process::drain_with_cap(data, 50).await;
617        assert_eq!(buf.len(), 50);
618        assert!(!truncated);
619    }
620}
camel_component_exec/producer.rs

camel_component_exec/
producer.rs
