molo_coding/coding/
command.rs

1use crate::harness::{ExecutionPolicy, NetworkPolicy, SandboxPolicy};
2use crate::{RunContext, RunMetadata};
3use async_trait::async_trait;
4use serde::{Deserialize, Serialize};
5use std::collections::BTreeMap;
6use std::fmt;
7use std::process::Stdio;
8use std::time::{Duration, Instant};
9use tokio::io::{AsyncReadExt, AsyncWriteExt};
10
11use super::workspace::{Workspace, WorkspacePath};
12
13/// Environment variable handling for command execution.
14#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
15#[non_exhaustive]
16pub enum EnvPolicy {
17    /// Start from an empty environment and use only `CommandRequest::env`.
18    #[default]
19    Empty,
20    /// Inherit only listed keys from the parent process, then apply
21    /// `CommandRequest::env`.
22    AllowList(Vec<String>),
23    /// Inherit the full parent environment. This is not the default because
24    /// it can leak secrets.
25    InheritAll,
26}
27
28/// PTY mode requested for a command.
29#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
30#[non_exhaustive]
31pub enum PtyMode {
32    /// No PTY. The local executor supports one-shot, non-interactive commands.
33    #[default]
34    Disabled,
35    /// Request a PTY. The local executor returns unsupported when PTY support
36    /// is unavailable.
37    Enabled,
38}
39
40/// Per-stream output limit.
41#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
42pub struct CommandOutputLimit {
43    /// Maximum stdout bytes returned.
44    pub stdout_bytes: usize,
45    /// Maximum stderr bytes returned.
46    pub stderr_bytes: usize,
47}
48
49impl Default for CommandOutputLimit {
50    fn default() -> Self {
51        Self {
52            stdout_bytes: 64 * 1024,
53            stderr_bytes: 64 * 1024,
54        }
55    }
56}
57
58/// Command execution request.
59#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
60pub struct CommandRequest {
61    /// Program and arguments. The first element is the program. No implicit
62    /// shell parsing is performed.
63    pub argv: Vec<String>,
64    /// Working directory inside the workspace.
65    pub cwd: WorkspacePath,
66    /// Explicit environment variables. Values are redacted from Debug.
67    pub env: BTreeMap<String, String>,
68    /// Parent environment inheritance policy.
69    pub env_policy: EnvPolicy,
70    /// Optional stdin bytes.
71    pub stdin: Option<Vec<u8>>,
72    /// PTY mode.
73    pub pty: PtyMode,
74    /// Request timeout.
75    pub timeout: Option<Duration>,
76    /// Requested sandbox policy.
77    pub requested_sandbox: Option<SandboxPolicy>,
78    /// Requested network policy.
79    pub requested_network: Option<NetworkPolicy>,
80    /// Output limits.
81    pub output_limit: CommandOutputLimit,
82    /// Host-owned metadata.
83    pub metadata: RunMetadata,
84}
85
86impl fmt::Debug for CommandRequest {
87    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
88        f.debug_struct("CommandRequest")
89            .field("argv", &self.argv)
90            .field("cwd", &self.cwd)
91            .field("env_keys", &self.env.keys().collect::<Vec<_>>())
92            .field("env_policy", &self.env_policy)
93            .field("stdin_len", &self.stdin.as_ref().map(Vec::len))
94            .field("pty", &self.pty)
95            .field("timeout", &self.timeout)
96            .field("requested_sandbox", &self.requested_sandbox)
97            .field("requested_network", &self.requested_network)
98            .field("output_limit", &self.output_limit)
99            .field("metadata", &self.metadata)
100            .finish()
101    }
102}
103
104impl CommandRequest {
105    /// Constructs a command request from argv.
106    pub fn new<I, S>(argv: I) -> Self
107    where
108        I: IntoIterator<Item = S>,
109        S: Into<String>,
110    {
111        Self {
112            argv: argv.into_iter().map(Into::into).collect(),
113            cwd: WorkspacePath::root(),
114            env: BTreeMap::new(),
115            env_policy: EnvPolicy::Empty,
116            stdin: None,
117            pty: PtyMode::Disabled,
118            timeout: None,
119            requested_sandbox: None,
120            requested_network: Some(NetworkPolicy::Deny),
121            output_limit: CommandOutputLimit::default(),
122            metadata: RunMetadata::new(),
123        }
124    }
125
126    /// Sets the working directory.
127    pub fn with_cwd(mut self, cwd: WorkspacePath) -> Self {
128        self.cwd = cwd;
129        self
130    }
131
132    /// Sets a timeout.
133    pub fn with_timeout(mut self, timeout: Duration) -> Self {
134        self.timeout = Some(timeout);
135        self
136    }
137}
138
139/// How strictly policy/capability mismatches are handled.
140#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
141#[non_exhaustive]
142pub enum PolicyCapabilityMode {
143    /// Required sandbox or network restrictions must be technically enforced.
144    #[default]
145    RequireEnforced,
146    /// Advisory execution may continue, but reports must say so explicitly.
147    AllowAdvisory,
148}
149
150impl PolicyCapabilityMode {
151    /// Returns true when advisory execution is explicitly allowed.
152    pub fn allows_advisory(self) -> bool {
153        matches!(self, Self::AllowAdvisory)
154    }
155}
156
157/// Structured enforcement status for a policy dimension.
158#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
159#[non_exhaustive]
160pub enum PolicyEnforcementStatus {
161    /// The requested policy was technically enforced by the executor backend.
162    Enforced,
163    /// The executor ran without technical enforcement and reported the downgrade.
164    Advisory,
165    /// The executor does not support this policy dimension.
166    Unsupported,
167    /// This policy dimension was not requested for the command.
168    NotRequested,
169    /// The executor could not determine whether enforcement happened.
170    Unknown,
171}
172
173impl PolicyEnforcementStatus {
174    /// Returns true when the status is [`PolicyEnforcementStatus::Enforced`].
175    pub fn is_enforced(self) -> bool {
176        matches!(self, Self::Enforced)
177    }
178
179    /// Returns true when the status is [`PolicyEnforcementStatus::Advisory`].
180    pub fn is_advisory(self) -> bool {
181        matches!(self, Self::Advisory)
182    }
183}
184
185/// Command executor backend family.
186#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
187#[non_exhaustive]
188pub enum CommandExecutorBackend {
189    /// Plain local process execution.
190    LocalProcess,
191    /// Host-provided backend outside molo's built-in executors.
192    HostProvided,
193    /// OS sandbox backend.
194    Sandbox,
195    /// Container backend.
196    Container,
197    /// Remote isolated worker backend.
198    Remote,
199    /// Application-specific backend kind.
200    Custom(String),
201}
202
203/// Executor identity included in capability and execution reports.
204#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
205pub struct CommandExecutorIdentity {
206    /// Executor name.
207    pub name: String,
208    /// Executor version, when available.
209    pub version: Option<String>,
210    /// Backend kind.
211    pub backend: CommandExecutorBackend,
212    /// Platform summary.
213    pub platform: String,
214    /// Host-owned metadata.
215    pub metadata: RunMetadata,
216}
217
218impl CommandExecutorIdentity {
219    /// Constructs an executor identity.
220    pub fn new(name: impl Into<String>, backend: CommandExecutorBackend) -> Self {
221        Self {
222            name: name.into(),
223            version: None,
224            backend,
225            platform: format!("{}/{}", std::env::consts::OS, std::env::consts::ARCH),
226            metadata: RunMetadata::new(),
227        }
228    }
229
230    /// Constructs the identity for [`LocalCommandExecutor`].
231    pub fn local_process() -> Self {
232        Self::new(
233            "local-command-executor",
234            CommandExecutorBackend::LocalProcess,
235        )
236        .with_version(env!("CARGO_PKG_VERSION"))
237    }
238
239    /// Sets executor version.
240    pub fn with_version(mut self, version: impl Into<String>) -> Self {
241        self.version = Some(version.into());
242        self
243    }
244
245    /// Sets host-owned metadata.
246    pub fn with_metadata(mut self, metadata: RunMetadata) -> Self {
247        self.metadata = metadata;
248        self
249    }
250}
251
252/// Executor capability report.
253#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
254pub struct CommandExecutorCapabilities {
255    /// Executor identity.
256    pub identity: CommandExecutorIdentity,
257    /// Whether non-PTY one-shot command execution is supported.
258    pub one_shot: bool,
259    /// Whether PTY command execution is supported.
260    pub pty: bool,
261    /// Whether the executor can technically enforce sandbox policy.
262    ///
263    /// Prefer [`CommandExecutorCapabilities::sandbox`] for new code.
264    pub sandbox_enforcement: bool,
265    /// Whether the executor can technically enforce network policy.
266    ///
267    /// Prefer [`CommandExecutorCapabilities::network`] for new code.
268    pub network_enforcement: bool,
269    /// Sandbox capability status for restrictive sandbox policies.
270    pub sandbox: PolicyEnforcementStatus,
271    /// Network capability status for restrictive network policies.
272    pub network: PolicyEnforcementStatus,
273    /// Whether timeout/cancellation can clean up the full process tree.
274    pub process_cleanup: PolicyEnforcementStatus,
275    /// Resource limit enforcement status beyond wall-time/output limits.
276    pub resource_limits: PolicyEnforcementStatus,
277    /// Host-owned capability metadata.
278    pub metadata: RunMetadata,
279}
280
281impl Default for CommandExecutorCapabilities {
282    fn default() -> Self {
283        Self::local_process()
284    }
285}
286
287impl CommandExecutorCapabilities {
288    /// Capability report for the built-in local executor.
289    pub fn local_process() -> Self {
290        Self {
291            identity: CommandExecutorIdentity::local_process(),
292            one_shot: true,
293            pty: false,
294            sandbox_enforcement: false,
295            network_enforcement: false,
296            sandbox: PolicyEnforcementStatus::Advisory,
297            network: PolicyEnforcementStatus::Advisory,
298            process_cleanup: PolicyEnforcementStatus::Advisory,
299            resource_limits: PolicyEnforcementStatus::Unsupported,
300            metadata: RunMetadata::new(),
301        }
302    }
303
304    /// Capability report for a host executor that enforces sandbox and network
305    /// restrictions.
306    pub fn enforced(identity: CommandExecutorIdentity) -> Self {
307        Self {
308            identity,
309            one_shot: true,
310            pty: false,
311            sandbox_enforcement: true,
312            network_enforcement: true,
313            sandbox: PolicyEnforcementStatus::Enforced,
314            network: PolicyEnforcementStatus::Enforced,
315            process_cleanup: PolicyEnforcementStatus::Enforced,
316            resource_limits: PolicyEnforcementStatus::Unknown,
317            metadata: RunMetadata::new(),
318        }
319    }
320}
321
322/// Text output with truncation metadata.
323#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
324pub struct OutputText {
325    /// UTF-8 lossy text.
326    pub text: String,
327    /// Original byte length.
328    pub bytes: usize,
329    /// Whether text was truncated.
330    pub truncated: bool,
331}
332
333/// Terminal command status.
334#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
335#[non_exhaustive]
336pub enum CommandStatus {
337    /// Process exited with an exit code.
338    Exited {
339        /// Exit code.
340        code: i32,
341    },
342    /// Process terminated by signal or platform-specific status.
343    Signaled {
344        /// Signal or platform-specific explanation.
345        signal: String,
346    },
347    /// Process exceeded timeout.
348    TimedOut,
349    /// Process was cancelled by run context.
350    Cancelled,
351}
352
353/// Report describing which policies were enforced by the command executor.
354#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
355pub struct PolicyEnforcementReport {
356    /// Executor identity.
357    pub executor: CommandExecutorIdentity,
358    /// Sandbox policy requested by the harness.
359    pub sandbox: SandboxPolicy,
360    /// Network policy requested by the harness.
361    pub network: NetworkPolicy,
362    /// Whether the sandbox policy was technically enforced.
363    pub sandbox_enforced: bool,
364    /// Whether the network policy was technically enforced.
365    pub network_enforced: bool,
366    /// Structured sandbox enforcement status.
367    pub sandbox_status: PolicyEnforcementStatus,
368    /// Structured network enforcement status.
369    pub network_status: PolicyEnforcementStatus,
370    /// Process tree cleanup status after timeout/cancellation.
371    pub process_cleanup_status: PolicyEnforcementStatus,
372    /// Resource limit enforcement status.
373    pub resource_limit_status: PolicyEnforcementStatus,
374    /// Warnings about advisory or unsupported policy.
375    pub warnings: Vec<String>,
376    /// Host-owned enforcement metadata.
377    pub metadata: RunMetadata,
378}
379
380/// Command output.
381#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
382pub struct CommandOutput {
383    /// Terminal status.
384    pub status: CommandStatus,
385    /// Captured stdout.
386    pub stdout: OutputText,
387    /// Captured stderr.
388    pub stderr: OutputText,
389    /// Process duration.
390    pub duration: Duration,
391    /// Whether either stream was truncated.
392    pub truncated: bool,
393    /// Policy enforcement report.
394    pub policy_enforcement: PolicyEnforcementReport,
395    /// Host-owned metadata.
396    pub metadata: RunMetadata,
397}
398
399/// Executes approved command requests.
400#[async_trait]
401pub trait CommandExecutor: Send + Sync {
402    /// Returns executor capabilities.
403    fn capabilities(&self) -> CommandExecutorCapabilities;
404
405    /// Executes a command under a harness execution policy.
406    async fn execute(
407        &self,
408        request: CommandRequest,
409        policy: &ExecutionPolicy,
410        context: &RunContext,
411    ) -> Result<CommandOutput, CommandError>;
412}
413
414/// Validates that executor capabilities can satisfy a command policy before
415/// execution starts.
416///
417/// # Errors
418///
419/// Returns [`CommandError::UnsupportedPolicy`] when required sandbox or
420/// network enforcement is missing and advisory mode is not enabled.
421pub fn validate_command_capabilities(
422    capabilities: &CommandExecutorCapabilities,
423    request: &CommandRequest,
424    policy: &ExecutionPolicy,
425    mode: PolicyCapabilityMode,
426) -> Result<(), CommandError> {
427    let sandbox = requested_sandbox(request, policy);
428    let network = requested_network(request, policy);
429    validate_required_status(
430        "sandbox",
431        sandbox_requires_enforcement(&sandbox),
432        capabilities.sandbox,
433        mode,
434    )?;
435    validate_required_status(
436        "network",
437        network_requires_enforcement(&network),
438        capabilities.network,
439        mode,
440    )
441}
442
443/// Validates that an executor's post-run report matches requested policy and
444/// declared capabilities.
445///
446/// # Errors
447///
448/// Returns [`CommandError::UnsupportedPolicy`] when the report downgrades or
449/// contradicts required enforcement.
450pub fn validate_policy_enforcement_report(
451    capabilities: &CommandExecutorCapabilities,
452    report: &PolicyEnforcementReport,
453    mode: PolicyCapabilityMode,
454) -> Result<(), CommandError> {
455    validate_report_status(
456        "sandbox",
457        sandbox_requires_enforcement(&report.sandbox),
458        capabilities.sandbox,
459        report.sandbox_status,
460        mode,
461    )?;
462    validate_report_status(
463        "network",
464        network_requires_enforcement(&report.network),
465        capabilities.network,
466        report.network_status,
467        mode,
468    )
469}
470
471/// Local non-PTY, one-shot command executor backed by host process spawning.
472///
473/// This executor resolves [`CommandRequest::cwd`] through the [`Workspace`],
474/// starts `argv` directly without implicit shell parsing, applies
475/// [`EnvPolicy`], requires a timeout, captures stdout/stderr separately, and
476/// reports output truncation.
477///
478/// This executor is not an OS sandbox. It does not technically enforce
479/// [`SandboxPolicy`], [`NetworkPolicy`], network isolation, process-tree
480/// cleanup, or resource limits. Its capability report marks sandbox, network,
481/// and process cleanup as advisory, and resource limits as unsupported.
482///
483/// By default, command execution fails closed when the requested policy requires
484/// technical enforcement that this local process backend cannot provide.
485/// [`LocalCommandExecutor::with_advisory_policy`] and
486/// [`LocalCommandExecutor::with_policy_capability_mode`] can explicitly allow
487/// such execution to continue with advisory enforcement reports. That mode is
488/// intended for tests, local prototypes, and reference CLI dogfooding; production
489/// coding-agent hosts should inject a [`CommandExecutor`] backed by a container,
490/// VM, platform sandbox, or remote isolated worker that can enforce the
491/// requested policy.
492#[derive(Debug, Clone)]
493pub struct LocalCommandExecutor<W> {
494    workspace: W,
495    policy_capability_mode: PolicyCapabilityMode,
496}
497
498impl<W> LocalCommandExecutor<W> {
499    /// Constructs a local command executor for a workspace.
500    pub fn new(workspace: W) -> Self {
501        Self {
502            workspace,
503            policy_capability_mode: PolicyCapabilityMode::RequireEnforced,
504        }
505    }
506
507    /// Allows policy requirements that cannot be technically enforced locally to
508    /// be reported as advisory instead of failing closed.
509    ///
510    /// Passing `true` does not enable sandboxing, network isolation, process-tree
511    /// cleanup, or resource limits. It only permits execution to continue while
512    /// recording the downgrade in the returned policy enforcement report.
513    pub fn with_advisory_policy(mut self, allow: bool) -> Self {
514        self.policy_capability_mode = if allow {
515            PolicyCapabilityMode::AllowAdvisory
516        } else {
517            PolicyCapabilityMode::RequireEnforced
518        };
519        self
520    }
521
522    /// Sets how this executor handles policy/capability mismatches.
523    ///
524    /// [`PolicyCapabilityMode::RequireEnforced`] is the conservative default.
525    /// [`PolicyCapabilityMode::AllowAdvisory`] allows local execution to proceed
526    /// without technical enforcement and requires reports to say so explicitly.
527    pub fn with_policy_capability_mode(mut self, mode: PolicyCapabilityMode) -> Self {
528        self.policy_capability_mode = mode;
529        self
530    }
531}
532
533#[async_trait]
534impl<W> CommandExecutor for LocalCommandExecutor<W>
535where
536    W: Workspace,
537{
538    fn capabilities(&self) -> CommandExecutorCapabilities {
539        CommandExecutorCapabilities::local_process()
540    }
541
542    async fn execute(
543        &self,
544        request: CommandRequest,
545        policy: &ExecutionPolicy,
546        context: &RunContext,
547    ) -> Result<CommandOutput, CommandError> {
548        if request.argv.is_empty() {
549            return Err(CommandError::InvalidRequest {
550                message: "argv must not be empty".to_string(),
551            });
552        }
553        if request.pty != PtyMode::Disabled {
554            return Err(CommandError::UnsupportedPolicy {
555                message: "PTY command execution is not supported by LocalCommandExecutor"
556                    .to_string(),
557            });
558        }
559        let sandbox = request
560            .requested_sandbox
561            .clone()
562            .unwrap_or_else(|| policy.sandbox().clone());
563        let network = request
564            .requested_network
565            .clone()
566            .unwrap_or_else(|| policy.network().clone());
567        let capabilities = self.capabilities();
568        validate_command_capabilities(
569            &capabilities,
570            &request,
571            policy,
572            self.policy_capability_mode,
573        )?;
574        let sandbox_status = requested_sandbox_status(&sandbox, capabilities.sandbox);
575        let network_status = requested_network_status(&network, capabilities.network);
576        let mut warnings = advisory_warnings(sandbox_status, network_status);
577
578        let cwd = self
579            .workspace
580            .resolve(&request.cwd, super::workspace::WorkspaceAccess::List)
581            .await
582            .map_err(|error| CommandError::InvalidRequest {
583                message: format!("invalid command cwd: {error}"),
584            })?;
585
586        let timeout = choose_timeout(request.timeout, policy.timeout(), context.remaining())
587            .ok_or_else(|| CommandError::InvalidRequest {
588                message: "command timeout is required".to_string(),
589            })?;
590        if timeout.is_zero() {
591            return Err(CommandError::TimedOut {
592                message: "command timeout elapsed".to_string(),
593            });
594        }
595
596        let mut command = tokio::process::Command::new(&request.argv[0]);
597        command.args(&request.argv[1..]);
598        command.current_dir(cwd.absolute);
599        command.stdin(if request.stdin.is_some() {
600            Stdio::piped()
601        } else {
602            Stdio::null()
603        });
604        command.stdout(Stdio::piped());
605        command.stderr(Stdio::piped());
606        command.kill_on_drop(true);
607        apply_env(&mut command, &request.env_policy, &request.env);
608
609        if let Some(stdin) = &request.stdin {
610            command.stdin(Stdio::piped());
611            command.env("MOLO_STDIN_BYTES", stdin.len().to_string());
612        }
613
614        let started = Instant::now();
615        let mut child = command.spawn().map_err(|error| CommandError::Spawn {
616            message: format!("failed to spawn command: {error}"),
617        })?;
618        if let Some(stdin) = &request.stdin
619            && let Some(mut child_stdin) = child.stdin.take()
620        {
621            child_stdin
622                .write_all(stdin)
623                .await
624                .map_err(|error| CommandError::Io {
625                    message: format!("failed to write stdin: {error}"),
626                })?;
627        }
628
629        let stdout_task = child.stdout.take().map(|mut stdout| {
630            tokio::spawn(async move {
631                let mut bytes = Vec::new();
632                stdout.read_to_end(&mut bytes).await.map(|_| bytes)
633            })
634        });
635        let stderr_task = child.stderr.take().map(|mut stderr| {
636            tokio::spawn(async move {
637                let mut bytes = Vec::new();
638                stderr.read_to_end(&mut bytes).await.map(|_| bytes)
639            })
640        });
641
642        let status = tokio::select! {
643            _ = context.cancellation.cancelled() => {
644                let _ = child.kill().await;
645                warnings.push("process tree cleanup is advisory for LocalCommandExecutor".to_string());
646                return Ok(terminal_output(
647                    CommandStatus::Cancelled,
648                    started.elapsed(),
649                    TerminalPolicyReportInput {
650                        executor: capabilities.identity.clone(),
651                        sandbox,
652                        network,
653                        sandbox_status,
654                        network_status,
655                        process_cleanup_status: PolicyEnforcementStatus::Advisory,
656                        warnings,
657                    },
658                    &request,
659                ));
660            }
661            result = tokio::time::timeout(timeout, child.wait()) => {
662                match result {
663                    Ok(status) => status.map_err(|error| CommandError::Io {
664                        message: format!("failed to wait for command: {error}"),
665                    })?,
666                    Err(_) => {
667                        let _ = child.kill().await;
668                        warnings.push("process tree cleanup is advisory for LocalCommandExecutor".to_string());
669                        return Ok(terminal_output(
670                            CommandStatus::TimedOut,
671                            started.elapsed(),
672                            TerminalPolicyReportInput {
673                                executor: capabilities.identity.clone(),
674                                sandbox,
675                                network,
676                                sandbox_status,
677                                network_status,
678                                process_cleanup_status: PolicyEnforcementStatus::Advisory,
679                                warnings,
680                            },
681                            &request,
682                        ));
683                    }
684                }
685            }
686        };
687
688        let stdout_bytes = join_reader(stdout_task).await?;
689        let stderr_bytes = join_reader(stderr_task).await?;
690        let stdout = truncate_bytes(&stdout_bytes, request.output_limit.stdout_bytes);
691        let stderr = truncate_bytes(&stderr_bytes, request.output_limit.stderr_bytes);
692        let truncated = stdout.truncated || stderr.truncated;
693        let status = match status.code() {
694            Some(code) => CommandStatus::Exited { code },
695            None => CommandStatus::Signaled {
696                signal: format!("{status:?}"),
697            },
698        };
699        Ok(CommandOutput {
700            status,
701            stdout,
702            stderr,
703            duration: started.elapsed(),
704            truncated,
705            policy_enforcement: PolicyEnforcementReport {
706                executor: capabilities.identity,
707                sandbox,
708                network,
709                sandbox_enforced: sandbox_status.is_enforced(),
710                network_enforced: network_status.is_enforced(),
711                sandbox_status,
712                network_status,
713                process_cleanup_status: PolicyEnforcementStatus::NotRequested,
714                resource_limit_status: PolicyEnforcementStatus::Unsupported,
715                warnings,
716                metadata: RunMetadata::new(),
717            },
718            metadata: command_metadata(&request),
719        })
720    }
721}
722
723fn apply_env(
724    command: &mut tokio::process::Command,
725    policy: &EnvPolicy,
726    env: &BTreeMap<String, String>,
727) {
728    command.env_clear();
729    match policy {
730        EnvPolicy::Empty => {}
731        EnvPolicy::AllowList(keys) => {
732            for key in keys {
733                if let Some(value) = std::env::var_os(key) {
734                    command.env(key, value);
735                }
736            }
737        }
738        EnvPolicy::InheritAll => {
739            for (key, value) in std::env::vars_os() {
740                command.env(key, value);
741            }
742        }
743    }
744    for (key, value) in env {
745        command.env(key, value);
746    }
747}
748
749fn choose_timeout(
750    request: Option<Duration>,
751    policy: Option<Duration>,
752    remaining: Option<Duration>,
753) -> Option<Duration> {
754    [request, policy, remaining]
755        .into_iter()
756        .flatten()
757        .reduce(|left, right| left.min(right))
758}
759
760fn requested_sandbox(request: &CommandRequest, policy: &ExecutionPolicy) -> SandboxPolicy {
761    request
762        .requested_sandbox
763        .clone()
764        .unwrap_or_else(|| policy.sandbox().clone())
765}
766
767fn requested_network(request: &CommandRequest, policy: &ExecutionPolicy) -> NetworkPolicy {
768    request
769        .requested_network
770        .clone()
771        .unwrap_or_else(|| policy.network().clone())
772}
773
774fn sandbox_requires_enforcement(sandbox: &SandboxPolicy) -> bool {
775    matches!(
776        sandbox,
777        SandboxPolicy::ReadOnly | SandboxPolicy::WorkspaceWrite | SandboxPolicy::Custom(_)
778    )
779}
780
781fn network_requires_enforcement(network: &NetworkPolicy) -> bool {
782    matches!(
783        network,
784        NetworkPolicy::Deny | NetworkPolicy::AllowListed(_) | NetworkPolicy::Custom(_)
785    )
786}
787
788fn requested_sandbox_status(
789    sandbox: &SandboxPolicy,
790    capability: PolicyEnforcementStatus,
791) -> PolicyEnforcementStatus {
792    if sandbox_requires_enforcement(sandbox) {
793        capability
794    } else {
795        PolicyEnforcementStatus::NotRequested
796    }
797}
798
799fn requested_network_status(
800    network: &NetworkPolicy,
801    capability: PolicyEnforcementStatus,
802) -> PolicyEnforcementStatus {
803    if network_requires_enforcement(network) {
804        capability
805    } else {
806        PolicyEnforcementStatus::NotRequested
807    }
808}
809
810fn validate_required_status(
811    dimension: &str,
812    requires_enforcement: bool,
813    status: PolicyEnforcementStatus,
814    mode: PolicyCapabilityMode,
815) -> Result<(), CommandError> {
816    if !requires_enforcement {
817        return Ok(());
818    }
819    match status {
820        PolicyEnforcementStatus::Enforced => Ok(()),
821        PolicyEnforcementStatus::Advisory if mode.allows_advisory() => Ok(()),
822        PolicyEnforcementStatus::Advisory => Err(CommandError::UnsupportedPolicy {
823            message: format!(
824                "{dimension} policy requires technical enforcement; executor only supports advisory mode"
825            ),
826        }),
827        PolicyEnforcementStatus::Unsupported => Err(CommandError::UnsupportedPolicy {
828            message: format!("{dimension} policy is unsupported by executor"),
829        }),
830        PolicyEnforcementStatus::NotRequested | PolicyEnforcementStatus::Unknown => {
831            Err(CommandError::UnsupportedPolicy {
832                message: format!("{dimension} policy enforcement status is {status:?}"),
833            })
834        }
835    }
836}
837
838fn validate_report_status(
839    dimension: &str,
840    requires_enforcement: bool,
841    capability: PolicyEnforcementStatus,
842    reported: PolicyEnforcementStatus,
843    mode: PolicyCapabilityMode,
844) -> Result<(), CommandError> {
845    if reported == PolicyEnforcementStatus::Enforced
846        && capability != PolicyEnforcementStatus::Enforced
847    {
848        return Err(CommandError::UnsupportedPolicy {
849            message: format!(
850                "{dimension} report claims enforced but capabilities report {capability:?}"
851            ),
852        });
853    }
854    if capability == PolicyEnforcementStatus::Enforced
855        && requires_enforcement
856        && reported != PolicyEnforcementStatus::Enforced
857    {
858        return Err(CommandError::UnsupportedPolicy {
859            message: format!(
860                "{dimension} capabilities require enforced report but executor returned {reported:?}"
861            ),
862        });
863    }
864    validate_required_status(dimension, requires_enforcement, reported, mode)
865}
866
867fn advisory_warnings(
868    sandbox_status: PolicyEnforcementStatus,
869    network_status: PolicyEnforcementStatus,
870) -> Vec<String> {
871    let mut warnings = Vec::new();
872    if sandbox_status.is_advisory() {
873        warnings.push("sandbox policy is advisory; no OS sandbox was applied".to_string());
874    }
875    if network_status.is_advisory() {
876        warnings.push("network policy is advisory; no network isolation was applied".to_string());
877    }
878    warnings
879}
880
881fn truncate_bytes(bytes: &[u8], max: usize) -> OutputText {
882    let truncated = bytes.len() > max;
883    let mut selected = bytes.to_vec();
884    if truncated {
885        selected.truncate(max);
886    }
887    OutputText {
888        text: String::from_utf8_lossy(&selected).into_owned(),
889        bytes: bytes.len(),
890        truncated,
891    }
892}
893
894async fn join_reader(
895    task: Option<tokio::task::JoinHandle<std::io::Result<Vec<u8>>>>,
896) -> Result<Vec<u8>, CommandError> {
897    let Some(task) = task else {
898        return Ok(Vec::new());
899    };
900    task.await
901        .map_err(|error| CommandError::Io {
902            message: format!("failed to join output reader: {error}"),
903        })?
904        .map_err(|error| CommandError::Io {
905            message: format!("failed to read command output: {error}"),
906        })
907}
908
909struct TerminalPolicyReportInput {
910    executor: CommandExecutorIdentity,
911    sandbox: SandboxPolicy,
912    network: NetworkPolicy,
913    sandbox_status: PolicyEnforcementStatus,
914    network_status: PolicyEnforcementStatus,
915    process_cleanup_status: PolicyEnforcementStatus,
916    warnings: Vec<String>,
917}
918
919fn terminal_output(
920    status: CommandStatus,
921    duration: Duration,
922    policy_report: TerminalPolicyReportInput,
923    request: &CommandRequest,
924) -> CommandOutput {
925    CommandOutput {
926        status,
927        stdout: truncate_bytes(&[], request.output_limit.stdout_bytes),
928        stderr: truncate_bytes(&[], request.output_limit.stderr_bytes),
929        duration,
930        truncated: false,
931        policy_enforcement: PolicyEnforcementReport {
932            executor: policy_report.executor,
933            sandbox: policy_report.sandbox,
934            network: policy_report.network,
935            sandbox_enforced: policy_report.sandbox_status.is_enforced(),
936            network_enforced: policy_report.network_status.is_enforced(),
937            sandbox_status: policy_report.sandbox_status,
938            network_status: policy_report.network_status,
939            process_cleanup_status: policy_report.process_cleanup_status,
940            resource_limit_status: PolicyEnforcementStatus::Unsupported,
941            warnings: policy_report.warnings,
942            metadata: RunMetadata::new(),
943        },
944        metadata: command_metadata(request),
945    }
946}
947
948fn command_metadata(request: &CommandRequest) -> RunMetadata {
949    let mut metadata = RunMetadata::new();
950    metadata.insert("argv".to_string(), serde_json::json!(request.argv));
951    metadata.insert(
952        "env_keys".to_string(),
953        serde_json::json!(request.env.keys().cloned().collect::<Vec<_>>()),
954    );
955    metadata
956}
957
958/// Command execution errors.
959#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error, Serialize, Deserialize)]
960#[non_exhaustive]
961pub enum CommandError {
962    /// Command request is invalid.
963    #[error("invalid command request: {message}")]
964    InvalidRequest {
965        /// Model-safe explanation.
966        message: String,
967    },
968    /// Requested policy cannot be enforced by this executor.
969    #[error("unsupported command policy: {message}")]
970    UnsupportedPolicy {
971        /// Model-safe explanation.
972        message: String,
973    },
974    /// Process spawn failed.
975    #[error("failed to spawn command: {message}")]
976    Spawn {
977        /// Model-safe explanation.
978        message: String,
979    },
980    /// Command timed out before producing output.
981    #[error("command timed out: {message}")]
982    TimedOut {
983        /// Model-safe explanation.
984        message: String,
985    },
986    /// Command was cancelled before producing output.
987    #[error("command cancelled: {message}")]
988    Cancelled {
989        /// Model-safe explanation.
990        message: String,
991    },
992    /// Output exceeded a hard limit.
993    #[error("command output limit: {message}")]
994    OutputLimit {
995        /// Model-safe explanation.
996        message: String,
997    },
998    /// I/O failed.
999    #[error("command I/O error: {message}")]
1000    Io {
1001        /// Model-safe explanation.
1002        message: String,
1003    },
1004}
1005
1006#[cfg(test)]
1007mod tests {
1008    use super::*;
1009    use crate::coding::LocalWorkspace;
1010
1011    fn temp_dir(tag: &str) -> std::path::PathBuf {
1012        let dir =
1013            std::env::temp_dir().join(format!("molo-command-test-{}-{tag}", std::process::id()));
1014        let _ = std::fs::remove_dir_all(&dir);
1015        std::fs::create_dir_all(&dir).unwrap();
1016        dir
1017    }
1018
1019    #[test]
1020    fn command_debug_redacts_env_values() {
1021        let mut request = CommandRequest::new(["echo", "ok"]);
1022        request
1023            .env
1024            .insert("TOKEN".to_string(), "secret".to_string());
1025        let debug = format!("{request:?}");
1026        assert!(debug.contains("TOKEN"));
1027        assert!(!debug.contains("secret"));
1028    }
1029
1030    #[tokio::test]
1031    async fn local_command_executes_without_shell_parsing() {
1032        let root = temp_dir("argv");
1033        let workspace = LocalWorkspace::new(&root).unwrap();
1034        let executor = LocalCommandExecutor::new(workspace).with_advisory_policy(true);
1035        let output = executor
1036            .execute(
1037                CommandRequest::new(["printf", "%s", "a;b"]),
1038                &ExecutionPolicy::new(SandboxPolicy::ReadOnly, NetworkPolicy::Deny)
1039                    .with_timeout(Some(Duration::from_secs(5))),
1040                &RunContext::new("cmd"),
1041            )
1042            .await
1043            .unwrap();
1044        assert_eq!(output.stdout.text, "a;b");
1045        assert_eq!(
1046            output.policy_enforcement.network_status,
1047            PolicyEnforcementStatus::Advisory
1048        );
1049        let _ = std::fs::remove_dir_all(root);
1050    }
1051
1052    #[tokio::test]
1053    async fn local_command_fails_closed_without_advisory_policy() {
1054        let root = temp_dir("fail-closed");
1055        let workspace = LocalWorkspace::new(&root).unwrap();
1056        let executor = LocalCommandExecutor::new(workspace);
1057        let error = executor
1058            .execute(
1059                CommandRequest::new(["printf", "ok"]),
1060                &ExecutionPolicy::new(SandboxPolicy::ReadOnly, NetworkPolicy::Deny)
1061                    .with_timeout(Some(Duration::from_secs(5))),
1062                &RunContext::new("cmd"),
1063            )
1064            .await
1065            .unwrap_err();
1066        assert!(matches!(error, CommandError::UnsupportedPolicy { .. }));
1067        let _ = std::fs::remove_dir_all(root);
1068    }
1069
1070    #[tokio::test]
1071    async fn local_command_truncates_stdout_and_stderr_separately() {
1072        let root = temp_dir("truncate");
1073        let workspace = LocalWorkspace::new(&root).unwrap();
1074        let executor = LocalCommandExecutor::new(workspace).with_advisory_policy(true);
1075        let mut request = CommandRequest::new(["sh", "-c", "printf 12345; printf abcde >&2"]);
1076        request.output_limit = CommandOutputLimit {
1077            stdout_bytes: 3,
1078            stderr_bytes: 2,
1079        };
1080        let output = executor
1081            .execute(
1082                request,
1083                &ExecutionPolicy::new(SandboxPolicy::ReadOnly, NetworkPolicy::Deny)
1084                    .with_timeout(Some(Duration::from_secs(5))),
1085                &RunContext::new("cmd"),
1086            )
1087            .await
1088            .unwrap();
1089        assert_eq!(output.stdout.text, "123");
1090        assert_eq!(output.stderr.text, "ab");
1091        assert!(output.truncated);
1092        let _ = std::fs::remove_dir_all(root);
1093    }
1094
1095    #[tokio::test]
1096    async fn local_command_truncation_handles_utf8_boundary() {
1097        let root = temp_dir("truncate-utf8");
1098        let workspace = LocalWorkspace::new(&root).unwrap();
1099        let executor = LocalCommandExecutor::new(workspace).with_advisory_policy(true);
1100        let mut request = CommandRequest::new(["printf", "éé"]);
1101        request.output_limit = CommandOutputLimit {
1102            stdout_bytes: 3,
1103            stderr_bytes: 3,
1104        };
1105        let output = executor
1106            .execute(
1107                request,
1108                &ExecutionPolicy::new(SandboxPolicy::ReadOnly, NetworkPolicy::Deny)
1109                    .with_timeout(Some(Duration::from_secs(5))),
1110                &RunContext::new("cmd"),
1111            )
1112            .await
1113            .unwrap();
1114
1115        assert_eq!(output.stdout.bytes, 4);
1116        assert!(output.stdout.truncated);
1117        assert!(!output.stdout.text.is_empty());
1118        let _ = std::fs::remove_dir_all(root);
1119    }
1120}
molo_coding/coding/command.rs

molo_coding/coding/
command.rs
