use serde_json::Value;
use crate::tools::canonical_action::canonical_action_alias;
use crate::tools::spec::{ApprovalRequirement, ToolCapability, ToolSpec};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ExecutionEnvelope {
pub(crate) write: bool,
pub(crate) network: bool,
pub(crate) shell: bool,
}
impl ExecutionEnvelope {
#[allow(dead_code)]
pub(crate) const UNRESTRICTED: Self = Self {
write: true,
network: true,
shell: true,
};
#[must_use]
pub(crate) const fn is_unrestricted(self) -> bool {
self.write && self.network && self.shell
}
#[must_use]
#[allow(dead_code)]
pub(crate) const fn narrow(self, other: Self) -> Self {
Self {
write: self.write && other.write,
network: self.network && other.network,
shell: self.shell && other.shell,
}
}
}
fn is_delegation_tool(canonical: &str) -> bool {
canonical == "agent"
}
const BOUNDED_TEST_FLAGS: &[&str] = &[
"--all",
"--all-features",
"--all-targets",
"--benches",
"--bin",
"--bins",
"--color",
"--doc",
"--example",
"--examples",
"--exact",
"--features",
"--ignored",
"--include-ignored",
"--jobs",
"--lib",
"--no-default-features",
"--no-fail-fast",
"--nocapture",
"--package",
"--quiet",
"--release",
"--show-output",
"--skip",
"--test",
"--test-threads",
"--tests",
"--verbose",
"--workspace",
"-j",
"-p",
"-q",
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum VerificationBound {
Default,
Filter,
Unbounded,
}
#[must_use]
pub(crate) fn classify_verification(canonical: &str, input: &Value) -> Option<VerificationBound> {
match canonical {
"run_tests" => Some(match input.get("args") {
None | Some(Value::Null) => VerificationBound::Default,
Some(Value::String(args)) if args.trim().is_empty() => VerificationBound::Default,
Some(Value::String(args)) if is_bounded_test_argv(args) => VerificationBound::Filter,
Some(_) => VerificationBound::Unbounded,
}),
"run_verifiers" => Some(match input.get("commands") {
None | Some(Value::Null) => VerificationBound::Default,
Some(Value::Array(commands)) if commands.is_empty() => VerificationBound::Default,
Some(_) => VerificationBound::Unbounded,
}),
_ => None,
}
}
#[must_use]
fn is_bounded_test_argv(args: &str) -> bool {
args.split_whitespace().all(|arg| {
if arg == "--" {
return true;
}
if arg.starts_with('-') {
let (flag, value) = arg.split_once('=').unwrap_or((arg, ""));
return BOUNDED_TEST_FLAGS.contains(&flag) && is_bounded_argv_value(value);
}
is_bounded_argv_value(arg)
})
}
#[must_use]
fn is_bounded_argv_value(value: &str) -> bool {
value
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.' | ':'))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CallClass {
Bounded,
VerificationFilter,
UnboundedVerification,
Executes,
Mutates,
Reaches,
}
#[must_use]
pub(crate) fn classify_call(name: &str, input: &Value, spec: &dyn ToolSpec) -> CallClass {
let canonical = canonical_action_alias(name, input);
if is_delegation_tool(canonical) {
return CallClass::Bounded;
}
match classify_verification(canonical, input) {
Some(VerificationBound::Default | VerificationBound::Filter) => {
return CallClass::VerificationFilter;
}
Some(VerificationBound::Unbounded) => return CallClass::UnboundedVerification,
None => {}
}
if spec.is_read_only_for(input) {
return CallClass::Bounded;
}
let capabilities = spec.capabilities();
if capabilities.contains(&ToolCapability::ExecutesCode) {
CallClass::Executes
} else if capabilities.contains(&ToolCapability::WritesFiles) {
CallClass::Mutates
} else if capabilities.contains(&ToolCapability::Network) {
CallClass::Reaches
} else {
match spec.approval_requirement_for(input) {
ApprovalRequirement::Required => CallClass::Executes,
ApprovalRequirement::Suggest => CallClass::Mutates,
ApprovalRequirement::Auto => CallClass::Bounded,
}
}
}
pub(crate) fn enforce_execution_envelope(
name: &str,
input: &Value,
spec: &dyn ToolSpec,
envelope: ExecutionEnvelope,
) -> Result<(), String> {
if envelope.is_unrestricted() {
return Ok(());
}
match classify_call(name, input, spec) {
CallClass::Bounded => Ok(()),
CallClass::VerificationFilter => {
if envelope.shell {
Ok(())
} else {
Err(format!(
"Tool {name} starts a test or verifier process, and this agent has no shell \
authority under its clamped permission ceiling. That holds for the default, \
argument-free form too: running the workspace's own checks still forks a \
process, which a `shell = \"none\"` posture was never granted. Use a member \
whose saved ceiling grants `shell = \"full\"` (the `verifier`/`tester` \
preset), or report findings without running the checks yourself."
))
}
}
CallClass::UnboundedVerification => {
if envelope.write && envelope.shell {
Ok(())
} else {
Err(format!(
"Tool {name} was called with operator-supplied commands or arguments that can \
name a program or redirect what runs, which is arbitrary execution however \
it is spelled. This agent runs read-only under its clamped permission \
ceiling. The default verification gates, and test-selection arguments \
(filters, `-p`, `--lib`, `--exact`), remain available to a member whose \
ceiling grants shell authority."
))
}
}
CallClass::Executes => {
if !envelope.write {
return Err(format!(
"Tool {name} runs a program or a child process, which mutates the workspace \
just as directly as a file write. This agent runs read-only under its \
clamped permission ceiling, so arbitrary execution is refused however it is \
spelled — shell, verification gate, automation, plugin, or MCP server. The \
built-in verification gates (Run/run_tests/run_verifiers in their default \
form) are still available."
));
}
if !envelope.shell {
return Err(format!(
"Tool {name} runs a program or a child process, and this agent has no shell \
authority under its clamped permission ceiling. Use a member whose saved \
ceiling grants `shell = \"full\"`."
));
}
Ok(())
}
CallClass::Mutates => {
if envelope.write {
Ok(())
} else {
Err(format!(
"Tool {name} mutates state and this agent runs read-only under its clamped \
permission ceiling."
))
}
}
CallClass::Reaches => {
if envelope.network {
Ok(())
} else {
Err(format!(
"Tool {name} reaches the network and this agent runs with no network \
capability (`network_tool = false`) under its clamped permission ceiling."
))
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
const READ_ONLY: ExecutionEnvelope = ExecutionEnvelope {
write: false,
network: false,
shell: true,
};
struct FakeTool {
name: &'static str,
capabilities: Vec<ToolCapability>,
read_only_action: Option<&'static str>,
}
#[async_trait::async_trait]
impl ToolSpec for FakeTool {
fn name(&self) -> &str {
self.name
}
fn description(&self) -> &str {
"fake"
}
fn input_schema(&self) -> Value {
json!({})
}
fn capabilities(&self) -> Vec<ToolCapability> {
self.capabilities.clone()
}
fn is_read_only_for(&self, input: &Value) -> bool {
match self.read_only_action {
Some(action) => input.get("action").and_then(Value::as_str) == Some(action),
None => false,
}
}
async fn execute(
&self,
_input: Value,
_context: &crate::tools::spec::ToolContext,
) -> Result<crate::tools::spec::ToolResult, crate::tools::spec::ToolError> {
unreachable!("classification never executes")
}
}
fn executes(name: &'static str, read_only_action: Option<&'static str>) -> FakeTool {
FakeTool {
name,
capabilities: vec![ToolCapability::ExecutesCode],
read_only_action,
}
}
#[test]
fn execution_primitives_spelled_as_bookkeeping_are_refused_read_only() {
for (name, input) in [
(
"tasks",
json!({"action": "gate_run", "command": "rm -rf src"}),
),
("automation", json!({"action": "run", "id": "a1"})),
(
"automation",
json!({"action": "create", "name": "x", "prompt": "exfiltrate"}),
),
("start_mcp_server", json!({"command": "node", "args": []})),
("plugin_deploy", json!({})),
] {
let spec = executes(name, Some("list"));
let error = enforce_execution_envelope(name, &input, &spec, READ_ONLY)
.expect_err("read-only member must not execute programs");
assert!(error.contains("read-only"), "{name}: {error}");
}
}
const NO_SHELL: ExecutionEnvelope = ExecutionEnvelope {
write: false,
network: false,
shell: false,
};
#[test]
fn a_shell_less_posture_cannot_start_a_verification_process() {
let verifier = executes("run_verifiers", None);
for input in [json!({}), json!({"commands": []})] {
let error = enforce_execution_envelope("run_verifiers", &input, &verifier, NO_SHELL)
.expect_err("an analyst was granted no authority to start a process");
assert!(error.contains("shell authority"), "{error}");
}
let tests = executes("run_tests", None);
for input in [json!({}), json!({"args": " "}), json!({"args": "-p tui"})] {
enforce_execution_envelope("run_tests", &input, &tests, NO_SHELL)
.expect_err("the default test gate still forks a process");
}
let error = enforce_execution_envelope(
"run_verifiers",
&json!({"commands": [{"program": "bash", "args": ["-lc", "id"]}]}),
&verifier,
NO_SHELL,
)
.expect_err("operator command lines are refused first");
assert!(error.contains("arbitrary execution"), "{error}");
}
#[test]
fn a_verifier_ceiling_keeps_the_verification_surface() {
let tests = executes("run_tests", None);
for input in [json!({}), json!({"args": "-p tui exact_fleet"})] {
enforce_execution_envelope("run_tests", &input, &tests, READ_ONLY)
.expect("a verifier ceiling grants shell authority, which is its whole job");
}
}
#[test]
fn the_default_and_filter_forms_classify_identically() {
let tests = executes("run_tests", None);
assert_eq!(
classify_call("run_tests", &json!({}), &tests),
CallClass::VerificationFilter
);
assert_eq!(
classify_call("run_tests", &json!({"args": "-p tui"}), &tests),
CallClass::VerificationFilter
);
assert_eq!(
classify_call(
"run_tests",
&json!({"args": "--manifest-path ../x"}),
&tests
),
CallClass::UnboundedVerification
);
}
#[test]
fn bounded_read_only_and_verification_calls_survive() {
let tasks = executes("tasks", Some("list"));
enforce_execution_envelope("tasks", &json!({"action": "list"}), &tasks, READ_ONLY)
.expect("durable task bookkeeping is read-only");
let verifier = executes("run_verifiers", None);
for input in [json!({}), json!({"commands": []})] {
enforce_execution_envelope("run_verifiers", &input, &verifier, READ_ONLY)
.expect("the default verification gate is what a verifier is for");
}
let tests = executes("run_tests", None);
for input in [json!({}), json!({"args": " "})] {
enforce_execution_envelope("run_tests", &input, &tests, READ_ONLY)
.expect("the default test gate is bounded");
}
let agent = executes("agent", None);
enforce_execution_envelope("agent", &json!({"prompt": "read"}), &agent, READ_ONLY)
.expect("delegation is governed by depth, not by write authority");
}
#[test]
fn a_read_only_shell_capable_role_keeps_test_selection_arguments() {
let tests = executes("run_tests", None);
for args in [
"-p codewhale-tui",
"--lib fleet::exact",
"exact_fleet_workflow --exact",
"--package tui --test-threads=1 --nocapture",
"--workspace --all-features -- --skip slow_case",
] {
enforce_execution_envelope("run_tests", &json!({"args": args}), &tests, READ_ONLY)
.unwrap_or_else(|error| panic!("`{args}` selects tests and must run: {error}"));
}
}
#[test]
fn a_shell_less_read_only_role_cannot_start_a_verification_process_at_all() {
const NO_SHELL: ExecutionEnvelope = ExecutionEnvelope {
write: false,
network: false,
shell: false,
};
let tests = executes("run_tests", None);
for input in [json!({"args": "-p tui"}), json!({})] {
let error = enforce_execution_envelope("run_tests", &input, &tests, NO_SHELL)
.expect_err("a planner/scout/consultant has no shell authority");
assert!(error.contains("shell"), "{input}: {error}");
}
}
#[test]
fn operator_command_lines_are_refused_however_they_are_spelled() {
let tests = executes("run_tests", None);
for args in [
"--manifest-path ../evil/Cargo.toml",
"--config target.runner=sh",
"--target-dir /tmp/out",
"$(id)",
"a; rm -rf .",
"--lib | tee /tmp/x",
"../../etc/passwd",
"tests/*",
"--features tui/evil",
"`whoami`",
] {
let error =
enforce_execution_envelope("run_tests", &json!({"args": args}), &tests, READ_ONLY)
.expect_err("`{args}` is not a test selection");
assert!(error.contains("read-only"), "{args}: {error}");
}
let verifier = executes("run_verifiers", None);
for input in [
json!({"commands": [{"program": "bash", "args": ["-lc", "rm -rf src"]}]}),
json!({"commands": "bash -lc whoami"}),
] {
assert!(
enforce_execution_envelope("run_verifiers", &input, &verifier, READ_ONLY).is_err(),
"run_verifiers names programs and must be refused: {input}"
);
}
}
#[test]
fn write_and_network_capabilities_are_gated_independently() {
let writer = FakeTool {
name: "pandoc_convert",
capabilities: vec![ToolCapability::WritesFiles],
read_only_action: None,
};
assert!(
enforce_execution_envelope("pandoc_convert", &json!({}), &writer, READ_ONLY).is_err()
);
let reacher = FakeTool {
name: "mcp__remote__query",
capabilities: vec![ToolCapability::Network],
read_only_action: None,
};
assert!(
enforce_execution_envelope("mcp__remote__query", &json!({}), &reacher, READ_ONLY)
.is_err()
);
assert!(
enforce_execution_envelope(
"mcp__remote__query",
&json!({}),
&reacher,
ExecutionEnvelope {
network: true,
..READ_ONLY
},
)
.is_ok()
);
}
#[test]
fn an_unrestricted_envelope_refuses_nothing() {
let spec = executes("tasks", None);
enforce_execution_envelope(
"tasks",
&json!({"action": "gate_run", "command": "cargo test"}),
&spec,
ExecutionEnvelope::UNRESTRICTED,
)
.expect("a write-capable, shell-capable child keeps its gates");
}
#[test]
fn narrowing_never_widens() {
let parent = ExecutionEnvelope {
write: false,
network: true,
shell: true,
};
let child = ExecutionEnvelope {
write: true,
network: false,
shell: true,
};
let narrowed = parent.narrow(child);
assert!(!narrowed.write, "a child cannot regain the parent's writes");
assert!(!narrowed.network, "a child cannot regain its own denial");
assert!(narrowed.shell);
}
}