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            rt.metrics().record_counter(
292                "exec_policy_denials_total",
293                1.0,
294                &[("reason", reason), ("route", &self.route_id)],
295            );
296        }
297    }
298
299    fn metric_outcome(
300        &self,
301        exit_code: Option<i32>,
302        timed_out: bool,
303        stdout_truncated: bool,
304        dur: Duration,
305    ) {
306        let Some(rt) = &self.rt else { return };
307        rt.metrics().record_histogram(
308            "exec_duration_secs",
309            dur.as_secs_f64(),
310            &[("route", &self.route_id)],
311        );
312        if let Some(c) = exit_code {
313            let code_str = c.to_string();
314            rt.metrics().record_counter(
315                "exec_exit_code",
316                1.0,
317                &[("code", &code_str), ("route", &self.route_id)],
318            );
319        }
320        if timed_out {
321            rt.metrics()
322                .record_counter("exec_timeouts_total", 1.0, &[("route", &self.route_id)]);
323        }
324        if stdout_truncated {
325            rt.metrics().record_counter(
326                "exec_stdout_truncated_total",
327                1.0,
328                &[("route", &self.route_id)],
329            );
330        }
331    }
332}
333
334/// Drain an Option<ChildStdout/Stderr> to (Bytes, truncated). Spawned as a task.
335async fn drain_join<R: tokio::io::AsyncRead + Unpin + Send + 'static>(
336    r: Option<R>,
337    cap: usize,
338) -> (Bytes, bool) {
339    match r {
340        Some(reader) => process::drain_with_cap(reader, cap).await,
341        None => (Bytes::new(), false),
342    }
343}
344
345/// Materialized result placed in the body. stdout/stderr are base64-encoded
346/// Strings (C-2: Bytes is not Serialize without a serde feature, and raw byte
347/// arrays would be a pathological JSON). `#[non_exhaustive]`.
348#[derive(Debug, serde::Serialize, serde::Deserialize)]
349#[non_exhaustive]
350pub struct ExecResult {
351    pub exit_code: Option<i32>,
352    pub stdout: String, // base64
353    pub stderr: String, // base64
354    pub stdout_truncated: bool,
355    pub stderr_truncated: bool,
356    pub timed_out: bool,
357    pub profile: String,
358    pub duration_ms: u64,
359}
360
361fn b64(b: &[u8]) -> String {
362    base64::engine::general_purpose::STANDARD.encode(b)
363}
364
365fn extract_stdin(ex: &Exchange, cap: usize) -> Result<Bytes, ExecError> {
366    let b = match &ex.input.body {
367        Body::Text(s) => Bytes::copy_from_slice(s.as_bytes()),
368        Body::Bytes(b) => Bytes::copy_from_slice(b),
369        Body::Json(v) => Bytes::from(serde_json::to_vec(v).unwrap_or_default()),
370        _ => Bytes::new(),
371    };
372    if b.len() > cap {
373        return Err(ExecError::StdinTooLarge {
374            size: b.len(),
375            max: cap,
376        });
377    }
378    Ok(b)
379}
380
381fn pack(route_id: &str, e: ExecError) -> CamelError {
382    CamelError::ProcessorErrorWithSource(format!("exec producer ({route_id}): {e}"), Arc::new(e))
383}
384
385#[cfg(test)]
386mod tests {
387    use super::*;
388    use crate::config::{ArgPolicy, ExecGlobalConfig, ExecProfile};
389    use camel_api::{Body, Exchange};
390    use std::collections::HashMap;
391    use std::path::PathBuf;
392    use std::sync::Arc;
393    use tokio::sync::Semaphore;
394
395    /// Minimal PATH lookup (mirrors config::which, but does NOT canonicalize).
396    /// Tests need the symlink name (e.g., "echo") preserved as argv[0] for multicall binaries.
397    fn which(name: &str) -> PathBuf {
398        let path = std::env::var_os("PATH").expect("PATH must be set");
399        for dir in std::env::split_paths(&path) {
400            let cand = dir.join(name);
401            if cand.is_file() {
402                return cand; // Do NOT canonicalize — preserve symlink name for argv[0]
403            }
404        }
405        panic!("{name} not found on PATH");
406    }
407
408    fn make_global() -> Arc<ExecGlobalConfig> {
409        let root = std::env::temp_dir().canonicalize().unwrap();
410        Arc::new(ExecGlobalConfig {
411            canonical_workspace_root: Some(root),
412            ..ExecGlobalConfig::default()
413        })
414    }
415
416    fn make_profile(
417        exe: &str,
418        args_policy: ArgPolicy,
419        accepted: Vec<i32>,
420        timeout: u64,
421    ) -> ExecProfile {
422        // Provide minimal env: set PATH explicitly so binaries can be found.
423        let mut env = crate::config::EnvPolicy::default();
424        if let Ok(path) = std::env::var("PATH") {
425            env.set.insert("PATH".into(), path);
426        }
427
428        ExecProfile {
429            name: "test".into(),
430            executable: exe.into(),
431            args: args_policy,
432            deny_flags: Default::default(),
433            allow_shell: true, // allow sh for tests
434            env,
435            working_dir: ".".into(),
436            timeout_secs: timeout,
437            accepted_exit_codes: accepted,
438            concurrency: 1,
439            stdin_max_bytes: 1 << 20,
440            stdout_max_bytes: 10 << 20,
441            stderr_max_bytes: 1 << 20,
442            sandbox: Default::default(),
443            output_mode: Default::default(),
444            canonical_executable: Some(which(exe)),
445        }
446    }
447
448    fn make_producer(profile: ExecProfile) -> ExecProducer {
449        ExecProducer {
450            profile: Arc::new(profile),
451            global: make_global(),
452            route_id: "test".into(),
453            host_env: Arc::new(HashMap::new()),
454            semaphore: Arc::new(Semaphore::new(1)),
455            rt: None,
456        }
457    }
458
459    /// Helper: set CamelExecArgs header on the exchange.
460    fn set_args(ex: &mut Exchange, args: Vec<&str>) {
461        let v: Vec<String> = args.into_iter().map(String::from).collect();
462        ex.input.headers.insert(
463            crate::headers::CAMEL_EXEC_ARGS.to_string(),
464            serde_json::to_value(v).unwrap(),
465        );
466    }
467
468    /// Helper: deserialize the ExecResult JSON from the exchange body.
469    fn result_from_body(ex: &Exchange) -> ExecResult {
470        let v = match &ex.input.body {
471            Body::Json(v) => v.clone(),
472            other => panic!("expected Body::Json, got {other:?}"),
473        };
474        serde_json::from_value(v).expect("deserialize ExecResult")
475    }
476
477    #[tokio::test]
478    async fn echo_happy_path() {
479        let profile = make_profile("echo", ArgPolicy::Any, vec![0], 30);
480        let producer = make_producer(profile);
481        let mut ex = Exchange::default();
482        set_args(&mut ex, vec!["hello"]);
483
484        producer.run(&mut ex).await.expect("echo must succeed");
485
486        let result = result_from_body(&ex);
487        assert_eq!(result.exit_code, Some(0));
488        assert!(!result.timed_out);
489        // stdout is base64-encoded "hello\n"
490        let stdout_bytes = base64::engine::general_purpose::STANDARD
491            .decode(&result.stdout)
492            .unwrap();
493        assert_eq!(stdout_bytes, b"hello\n");
494        // Header assertions
495        assert_eq!(
496            ex.input
497                .headers
498                .get(crate::headers::CAMEL_EXEC_EXIT_ACCEPTED),
499            Some(&serde_json::Value::Bool(true))
500        );
501        assert_eq!(
502            ex.input.headers.get(crate::headers::CAMEL_EXEC_TIMED_OUT),
503            Some(&serde_json::Value::Bool(false))
504        );
505    }
506
507    #[tokio::test]
508    async fn timeout_is_non_error_outcome() {
509        let profile = make_profile("sleep", ArgPolicy::Any, vec![0], 1);
510        let producer = make_producer(profile);
511        let mut ex = Exchange::default();
512        set_args(&mut ex, vec!["30"]);
513
514        // KEY ASSERTION: timeout returns Ok(()), NOT Err.
515        let res = producer.run(&mut ex).await;
516        assert!(
517            res.is_ok(),
518            "timeout must be non-error outcome, got {res:?}"
519        );
520
521        let result = result_from_body(&ex);
522        assert_eq!(result.exit_code, None, "killed process has no exit code");
523        assert!(result.timed_out, "must be marked as timed out");
524        // CamelExecExitAccepted must be false when no exit code
525        assert_eq!(
526            ex.input
527                .headers
528                .get(crate::headers::CAMEL_EXEC_EXIT_ACCEPTED),
529            Some(&serde_json::Value::Bool(false))
530        );
531        assert_eq!(
532            ex.input.headers.get(crate::headers::CAMEL_EXEC_TIMED_OUT),
533            Some(&serde_json::Value::Bool(true))
534        );
535    }
536
537    #[tokio::test]
538    async fn non_zero_exit_is_non_error_outcome() {
539        // Case 1: accepted_exit_codes=[0], actual exit=2 → accepted=false
540        let profile = make_profile("sh", ArgPolicy::Any, vec![0], 30);
541        let producer = make_producer(profile);
542        let mut ex = Exchange::default();
543        set_args(&mut ex, vec!["-c", "exit 2"]);
544
545        let res = producer.run(&mut ex).await;
546        assert!(res.is_ok(), "non-zero exit must be non-error outcome");
547
548        let result = result_from_body(&ex);
549        assert_eq!(result.exit_code, Some(2));
550        assert!(!result.timed_out);
551        assert_eq!(
552            ex.input
553                .headers
554                .get(crate::headers::CAMEL_EXEC_EXIT_ACCEPTED),
555            Some(&serde_json::Value::Bool(false)),
556            "exit 2 not in accepted_exit_codes=[0]"
557        );
558
559        // Case 2: accepted_exit_codes=[2], actual exit=2 → accepted=true
560        let profile2 = make_profile("sh", ArgPolicy::Any, vec![2], 30);
561        let producer2 = make_producer(profile2);
562        let mut ex2 = Exchange::default();
563        set_args(&mut ex2, vec!["-c", "exit 2"]);
564
565        producer2.run(&mut ex2).await.expect("must succeed");
566        assert_eq!(
567            ex2.input
568                .headers
569                .get(crate::headers::CAMEL_EXEC_EXIT_ACCEPTED),
570            Some(&serde_json::Value::Bool(true)),
571            "exit 2 in accepted_exit_codes=[2]"
572        );
573    }
574
575    #[test]
576    fn clone_shares_host_env() {
577        let producer = make_producer(make_profile("echo", ArgPolicy::Any, vec![0], 30));
578        let cloned = producer.clone();
579        assert!(
580            Arc::ptr_eq(&producer.host_env, &cloned.host_env),
581            "clone must share host_env Arc, not deep-copy"
582        );
583    }
584
585    #[tokio::test]
586    async fn drain_with_cap_stores_prefix_on_overflow() {
587        // &[u8] delivers all 100 bytes in a single read (tmp buf is 8192),
588        // so this exercises the n > space branch (store space, flip truncated).
589        let data: &[u8] = &[b'x'; 100];
590        let (buf, truncated) = process::drain_with_cap(data, 50).await;
591        assert_eq!(buf.len(), 50, "buf must contain exactly cap bytes");
592        assert!(truncated);
593    }
594
595    #[tokio::test]
596    async fn drain_with_cap_multi_read_boundary() {
597        // Chain two slices to force two reads: 30 bytes then 30 bytes.
598        // First read fits (buf=30), second read overflows (space=20 → store 20, flip).
599        use tokio::io::AsyncReadExt;
600        let r1: &[u8] = &[b'x'; 30];
601        let r2: &[u8] = &[b'y'; 30];
602        let chained = r1.chain(r2);
603        let (buf, truncated) = process::drain_with_cap(chained, 50).await;
604        assert_eq!(buf.len(), 50);
605        assert!(truncated);
606    }
607
608    #[tokio::test]
609    async fn drain_with_cap_exact_fit_not_truncated() {
610        let data: &[u8] = &[b'x'; 50];
611        let (buf, truncated) = process::drain_with_cap(data, 50).await;
612        assert_eq!(buf.len(), 50);
613        assert!(!truncated);
614    }
615}
camel_component_exec/producer.rs

camel_component_exec/
producer.rs
