use std::collections::BTreeMap;
use super::*;
use crate::orchestration::{
pop_execution_policy, push_execution_policy, CapabilityPolicy, ExternalRoot,
};
use crate::tool_annotations::{SideEffectLevel, ToolAnnotations, ToolArgSchema, ToolKind};
#[path = "authority_acquisition_tests.rs"]
mod authority_acquisition;
fn policy_with_path_annotation(tool: &str, kind: ToolKind) {
let mut annotations = BTreeMap::new();
annotations.insert(
tool.to_string(),
ToolAnnotations {
kind,
side_effect_level: match kind {
ToolKind::Fetch => SideEffectLevel::Network,
ToolKind::Execute => SideEffectLevel::ProcessExec,
ToolKind::Edit | ToolKind::Delete | ToolKind::Move => {
SideEffectLevel::WorkspaceWrite
}
_ => SideEffectLevel::ReadOnly,
},
arg_schema: ToolArgSchema {
path_params: vec!["path".to_string()],
..Default::default()
},
..Default::default()
},
);
push_execution_policy(CapabilityPolicy {
tool_annotations: annotations,
..Default::default()
});
}
#[test]
fn compact_rule_shorthand_deserializes() {
let rule: PolicyRule = serde_json::from_value(serde_json::json!({
"deny": {"tool": "read_*", "path": "**/.env"},
"reason": "secret file"
}))
.expect("rule");
assert_eq!(rule.action, PolicyAction::Deny);
assert_eq!(rule.matches.tool, vec!["read_*"]);
assert_eq!(rule.matches.path, vec!["**/.env"]);
assert_eq!(rule.reason.as_deref(), Some("secret file"));
}
#[test]
fn ambiguous_or_invalid_rule_shapes_are_rejected() {
let invalid_action = serde_json::from_value::<PolicyRule>(serde_json::json!({
"action": "maybe",
"match": {"tool": "read_file"}
}));
assert!(invalid_action.is_err());
let mixed_matchers = serde_json::from_value::<PolicyRule>(serde_json::json!({
"action": "deny",
"match": {"tool": "read_file"},
"path": "**/.env"
}));
assert!(mixed_matchers.is_err());
let invalid_source = serde_json::from_value::<PolicyRule>(serde_json::json!({
"source": "superuser",
"allow": {"tool": "read_file"}
}));
assert!(invalid_source.is_err());
}
#[test]
fn deny_beats_ask_and_allow_regardless_of_order() {
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [
{"allow": {"tool": "write_file"}},
{"ask": {"tool": "write_*"}},
{"deny": {"tool": "write_file"}, "reason": "blocked"}
]
}))
.expect("policy");
let decision =
evaluate_tool_approval_policy(&policy, "write_file", &serde_json::json!({}), None);
assert!(decision.is_deny());
assert_eq!(decision.reason, "blocked");
assert_eq!(
decision.matched_rule.as_ref().and_then(|rule| rule.index),
Some(2)
);
}
#[test]
fn explicit_user_rules_outrank_mode_defaults_and_name_the_winner() {
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [
{"id": "mode-ask", "source": "mode", "ask": {"tool": "run_command"}},
{"id": "remembered-allow", "source": "user", "allow": {"tool": "run_command"}}
]
}))
.expect("policy with typed user source");
let decision = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"command": "git status"}),
None,
);
assert!(decision.is_allow(), "{decision:?}");
assert_eq!(
decision
.matched_rule
.as_ref()
.map(|rule| rule.source.as_str()),
Some("user")
);
assert_eq!(decision.receipt["matched_rule"]["source"], "user");
let host_decision = policy.evaluate_request(&ToolApprovalRequest {
tool_name: "run_command".to_string(),
arguments: serde_json::json!({"command": "git status"}),
..Default::default()
});
assert_eq!(host_decision.action, decision.action);
assert_eq!(host_decision.receipt["matched_rule"]["source"], "user");
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [
{"id": "mode-allow", "source": "mode", "allow": {"tool": "run_command"}},
{"id": "remembered-deny", "source": "user", "deny": {"tool": "run_command"}}
]
}))
.expect("policy with typed user source");
let decision = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"command": "git status"}),
None,
);
assert!(decision.is_deny(), "{decision:?}");
assert_eq!(
decision
.matched_rule
.as_ref()
.map(|rule| rule.source.as_str()),
Some("user")
);
}
#[test]
fn user_allow_cannot_override_path_guards() {
policy_with_path_annotation("read_file", ToolKind::Read);
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [{"source": "user", "allow": {"tool": "read_file"}}]
}))
.expect("policy with typed user source");
for (path, source) in [
("config/.env", SOURCE_DEFAULT_SENSITIVE_PATH),
("/tmp/outside.txt", SOURCE_DEFAULT_EXTERNAL_PATH),
("../outside.txt", SOURCE_DEFAULT_PATH_GUARD),
] {
let decision = evaluate_tool_approval_policy(
&policy,
"read_file",
&serde_json::json!({"path": path}),
None,
);
assert!(decision.is_deny(), "{path}: {decision:?}");
assert_eq!(
decision
.matched_rule
.as_ref()
.map(|rule| rule.source.as_str()),
Some(source)
);
}
pop_execution_policy();
}
#[test]
fn source_priority_only_overrides_mode_defaults() {
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"require_approval": ["run_command"],
"rules": [
{"source": "user", "allow": {"tool": "run_command"}, "id": "user-allow"}
]
}))
.expect("legacy approval constraint");
let decision = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"command": "git status"}),
None,
);
assert!(decision.is_ask(), "{decision:?}");
assert_eq!(
decision
.matched_rule
.as_ref()
.map(|rule| rule.source.as_str()),
Some("require_approval")
);
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [
{"source": "mode", "deny": {"tool": "run_command"}, "id": "mode-deny"},
{"source": "user", "allow": {"tool": "run_command"}, "id": "user-allow"}
]
}))
.expect("typed sources");
let decision = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"command": "git status"}),
None,
);
assert!(decision.is_allow(), "{decision:?}");
assert_eq!(
decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("user-allow")
);
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [
{"source": "mode", "deny": {"tool": "run_command"}, "id": "mode-deny"},
{"source": "user", "allow": {"tool": "run_command"}, "id": "user-allow"},
{"source": "policy", "deny": {"tool": "run_command"}, "id": "policy-deny"}
]
}))
.expect("typed sources");
let decision = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"command": "git status"}),
None,
);
assert!(decision.is_deny(), "{decision:?}");
assert_eq!(
decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("policy-deny")
);
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [
{"source": "policy", "allow": {"tool": "run_command"}, "id": "policy-allow"},
{"source": "user", "deny": {"tool": "run_command"}, "id": "user-deny"}
]
}))
.expect("a policy allow and explicit user deny");
let decision = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"command": "git status"}),
None,
);
assert!(decision.is_deny(), "{decision:?}");
assert_eq!(
decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("user-deny")
);
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [
{"source": "user", "allow": {"tool": "run_command"}, "id": "first-allow"},
{"source": "user", "deny": {"tool": "run_command"}, "id": "second-deny"}
]
}))
.expect("same-source rules");
let decision = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"command": "git status"}),
None,
);
assert!(decision.is_deny(), "{decision:?}");
assert_eq!(
decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("second-deny")
);
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"rules": [
{"source": "user", "allow": {"tool": "run_command"}, "id": "first"},
{"source": "user", "allow": {"tool": "run_command"}, "id": "second"}
]
}))
.expect("same-source ties");
let decision = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"command": "git status"}),
None,
);
assert_eq!(
decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("first")
);
}
#[test]
fn sensitive_paths_are_denied_by_default() {
policy_with_path_annotation("read_file", ToolKind::Read);
let policy = ToolApprovalPolicy::default();
let decisions = [
"config/.env",
".env.local",
"/home/agent/.ssh/id_rsa",
"/home/agent/.aws/credentials",
"certificates/client.pem",
"certificates/client.key",
]
.map(|path| {
evaluate_tool_approval_policy(
&policy,
"read_file",
&serde_json::json!({"path": path}),
None,
)
});
pop_execution_policy();
for decision in decisions {
assert!(decision.is_deny(), "sensitive path was allowed");
assert!(decision.risk_labels.contains(&"sensitive_path".to_string()));
}
}
#[test]
fn unannotated_approval_api_preserves_conventional_path_safety() {
let policy = ToolApprovalPolicy::default();
assert!(evaluate_tool_approval_policy(
&policy,
"read_file",
&serde_json::json!({"path": ".env"}),
None,
)
.is_deny());
assert!(evaluate_tool_approval_policy(
&policy,
"render",
&serde_json::json!({"content": "document cat .env here"}),
None,
)
.is_allow());
}
#[test]
fn sensitive_command_reads_use_structural_command_facts() {
let policy = ToolApprovalPolicy::default();
for arguments in [
serde_json::json!({"argv": ["cat", ".env"]}),
serde_json::json!({"command": "sh -c 'cat .env'"}),
] {
assert!(
evaluate_tool_approval_policy(&policy, "run_command", &arguments, None).is_deny(),
"sensitive command read was allowed: {arguments}"
);
}
let payload = serde_json::json!({"argv": ["python", "-c", "print('cat .env')"]});
assert!(
evaluate_tool_approval_policy(&policy, "run_command", &payload, None).is_allow(),
"inert interpreter payload was treated as a command"
);
}
#[test]
fn sensitive_path_default_does_not_classify_edit_content_as_a_path() {
policy_with_path_annotation("edit_file", ToolKind::Edit);
let policy = ToolApprovalPolicy::default();
let decisions = [
"value = os.environ['API_KEY']\n",
"const value = process.env.API_KEY;\n",
"for key in config.keys():\n print(key)\n",
"documentation = 'read config/.env before starting'\n",
]
.map(|content| {
evaluate_tool_approval_policy(
&policy,
"edit_file",
&serde_json::json!({"path": "src/example.py", "content": content}),
None,
)
});
pop_execution_policy();
for decision in decisions {
assert!(
decision.is_allow(),
"non-path content was denied: {}",
decision.reason
);
}
}
#[test]
fn sensitive_path_default_uses_path_globs_not_substrings() {
policy_with_path_annotation("read_file", ToolKind::Read);
let policy = ToolApprovalPolicy::default();
let decisions = ["src/os.environment.rs", "src/thing.keyword"].map(|path| {
evaluate_tool_approval_policy(
&policy,
"read_file",
&serde_json::json!({"path": path}),
None,
)
});
pop_execution_policy();
for decision in decisions {
assert!(
decision.is_allow(),
"lookalike path was denied: {}",
decision.reason
);
}
}
#[test]
fn sensitive_path_denial_reason_has_bounded_evidence() {
policy_with_path_annotation("read_file", ToolKind::Read);
let long_path = format!("{}/.env", "directory".repeat(128));
let decision = evaluate_tool_approval_policy(
&ToolApprovalPolicy::default(),
"read_file",
&serde_json::json!({"path": long_path}),
None,
);
pop_execution_policy();
assert!(decision.is_deny());
assert!(decision.reason.chars().count() < 300, "{}", decision.reason);
assert!(decision.reason.contains('…'), "{}", decision.reason);
assert!(decision.reason.contains("/.env"), "{}", decision.reason);
}
#[test]
fn custom_sensitive_path_patterns_preserve_full_glob_grammar() {
let policy = ToolApprovalPolicy {
sensitive_path_patterns: vec![
"secret[0-9].txt".to_string(),
"config/{prod,stage}/credentials.json".to_string(),
],
..Default::default()
};
for path in ["secret7.txt", "config/prod/credentials.json"] {
assert!(
evaluate_tool_approval_policy(
&policy,
"read_file",
&serde_json::json!({"path": path}),
None,
)
.is_deny(),
"custom sensitive glob did not match {path}"
);
}
}
#[test]
fn explicit_sensitive_opt_out_allows_regular_evaluation() {
let policy = ToolApprovalPolicy {
allow_sensitive_paths: true,
..Default::default()
};
let decision = evaluate_tool_approval_policy(
&policy,
"read_file",
&serde_json::json!({"path": "config/.env"}),
None,
);
assert!(decision.is_allow());
assert!(!decision.has_audit_signal());
let command = evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"argv": ["cat", ".env"]}),
None,
);
assert!(command.is_allow());
}
#[test]
fn external_declared_paths_are_denied_without_root() {
let temp = tempfile::tempdir().unwrap();
std::fs::create_dir_all(temp.path().join("packages/app")).unwrap();
std::fs::write(temp.path().join("packages/app/host.harn"), "ok").unwrap();
crate::stdlib::process::set_thread_execution_context(Some(
crate::orchestration::RunExecutionRecord {
cwd: Some(temp.path().to_string_lossy().into_owned()),
project_root: None,
source_dir: Some(temp.path().to_string_lossy().into_owned()),
env: BTreeMap::new(),
adapter: None,
repo_path: None,
worktree_path: None,
branch: None,
base_ref: None,
cleanup: None,
environment_policy: Default::default(),
grants: Vec::new(),
admitted_environment: Vec::new(),
},
));
policy_with_path_annotation("read_file", ToolKind::Read);
for path in ["/tmp/outside.txt", "/packages/app/host.harn"] {
let decision = evaluate_tool_approval_policy(
&ToolApprovalPolicy::default(),
"read_file",
&serde_json::json!({"path": path}),
None,
);
assert!(decision.is_deny(), "{path}: {decision:?}");
assert!(decision.risk_labels.contains(&"external_path".to_string()));
assert_eq!(
decision.receipt["matched_rule"]["source"],
"default_external_path"
);
}
for path in [
"packages/app/host.harn".to_string(),
temp.path()
.join("packages/app/host.harn")
.to_string_lossy()
.into_owned(),
] {
let decision = evaluate_tool_approval_policy(
&ToolApprovalPolicy::default(),
"read_file",
&serde_json::json!({"path": path}),
None,
);
assert!(decision.is_allow(), "{path}: {decision:?}");
}
pop_execution_policy();
crate::stdlib::process::set_thread_execution_context(None);
}
#[test]
fn path_rule_uses_declared_path_params() {
policy_with_path_annotation("write_file", ToolKind::Edit);
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [{"ask": {"tool": "write_*", "path": "src/**"}, "reason": "source edit"}]
}))
.expect("policy");
let decision = evaluate_tool_approval_policy(
&policy,
"write_file",
&serde_json::json!({"path": "src/lib.rs"}),
None,
);
assert!(decision.is_ask());
assert_eq!(decision.reason, "source edit");
pop_execution_policy();
}
#[test]
fn command_url_mcp_identity_and_repeat_rules_match() {
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [
{"ask": {"tool": "run_command", "command_identity": "npm"}},
{"deny": {"tool": "fetch_url", "domain": "*.example.com", "method": "POST"}},
{"deny": {"mcp_server": "github", "mcp_tool": "create_issue"}},
{"deny": {"tool": "read_file", "repeat_count_gte": 3}}
]
}))
.expect("policy");
assert!(evaluate_tool_approval_policy(
&policy,
"run_command",
&serde_json::json!({"argv": ["npm", "install"]}),
None,
)
.is_ask());
assert!(evaluate_tool_approval_policy(
&policy,
"fetch_url",
&serde_json::json!({"url": "https://api.example.com/v1", "method": "post"}),
None,
)
.is_deny());
assert!(evaluate_tool_approval_policy(
&policy,
"github__create_issue",
&serde_json::json!({}),
None,
)
.is_deny());
assert!(evaluate_tool_approval_policy(
&policy,
"read_file",
&serde_json::json!({"path": "README.md"}),
Some(3),
)
.is_deny());
}
#[test]
fn persona_agent_and_mode_rules_match_args() {
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [{"deny": {"agent": "release-*", "persona": "shipper", "mode": "act"}}]
}))
.expect("policy");
let decision = evaluate_tool_approval_policy(
&policy,
"publish",
&serde_json::json!({"agent": "release-1", "persona": "shipper", "mode": "act"}),
None,
);
assert!(decision.is_deny());
}
#[test]
fn host_request_normalizes_harn_context_and_emits_canonical_receipt() {
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [
{
"id": "allow-fallback",
"allow": {"tool": "run"}
},
{
"id": "deny-release-publish",
"deny": {
"tool": "run",
"tool_kind": "execute",
"side_effect": "process_exec",
"command_identity": "cargo",
"domain": "crates.io",
"method": "POST",
"agent": "release-*",
"persona": "shipper",
"mode": "act",
"env_mode": "patch",
"capability": "terminal.execute"
},
"reason": "release publishing is supervised"
}
]
}))
.expect("policy");
let request = ToolApprovalRequest {
tool_name: "run".to_string(),
arguments: serde_json::json!({
"rawInput": {
"argv": ["cargo", "publish"],
"url": "https://crates.io/api/v1/crates",
"method": "post",
"envMode": "patch"
}
}),
policy_decision: Some(serde_json::json!({
"context": {
"toolName": "run",
"toolKind": "execute",
"requestedSideEffectLevel": "process_exec",
"agent_id": "release-1",
"persona_id": "shipper",
"action": "act",
"policy_context": {
"capabilities": ["terminal.execute"]
}
}
})),
approval_request: None,
repeat_count: None,
..Default::default()
};
let decision = policy.evaluate_request(&request);
assert!(decision.is_deny());
assert_eq!(decision.reason, "release publishing is supervised");
assert_eq!(
decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("deny-release-publish")
);
assert_eq!(
decision.receipt.get("type").and_then(JsonValue::as_str),
Some(POLICY_RECEIPT_TYPE)
);
assert_eq!(
decision
.receipt
.pointer("/context/command_identities/0")
.and_then(JsonValue::as_str),
Some("cargo")
);
}
#[test]
fn host_request_normalizes_generic_argument_paths() {
let workspace = tempfile::tempdir().unwrap();
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [{
"id": "ask-source-read",
"ask": {"tool": "read", "path": "src/**"},
"reason": "source read"
}]
}))
.expect("policy");
let request = ToolApprovalRequest {
tool_name: "read".to_string(),
arguments: serde_json::json!({"path": "src/main.rs"}),
workspace_boundary: Some(ToolApprovalWorkspaceBoundary {
root: workspace.path().to_str().unwrap().into(),
}),
..Default::default()
};
let decision = policy.evaluate_request(&request);
assert!(decision.is_ask());
assert_eq!(decision.reason, "source read");
assert_eq!(
decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("ask-source-read")
);
}
#[test]
fn host_request_declared_schema_owns_paths_and_tool_effects() {
let workspace = tempfile::tempdir().unwrap();
let annotations = ToolAnnotations {
kind: ToolKind::Edit,
side_effect_level: SideEffectLevel::WorkspaceWrite,
arg_schema: ToolArgSchema {
path_params: vec!["filename".into()],
..Default::default()
},
..Default::default()
};
let request = ToolApprovalRequest::from_host_json(serde_json::json!({
"tool_name": "custom_edit",
"arguments": {"filename": "actual.txt", "target": {"presentation": "not a path"}},
"workspace_boundary": {"root": workspace.path().to_str().unwrap()},
"tool_annotations": annotations,
"policy_decision": {"context": {
"tool_kind": "read", "side_effect": "read_only",
"paths": [{"input": "remembered.txt", "workspace_path": "remembered.txt"}]
}}
}))
.unwrap();
let policy = ToolApprovalPolicy::from_host_json(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [
{"id": "actual", "source": "user", "identity_match": "literal", "allow": {
"tool": "custom_edit", "tool_kind": "edit", "side_effect": "workspace_write", "path": "actual.txt"
}},
{"id": "ask", "source": "mode", "ask": {"tool": "custom_edit"}}
]
})).unwrap();
let decision = policy.evaluate_request(&request);
assert!(decision.is_allow(), "{decision:?}");
assert_eq!(decision.receipt["matched_rule"]["id"], "actual");
assert_eq!(decision.receipt["context"]["tool_kind"], "edit");
let mut different = request;
different.arguments["filename"] = serde_json::json!("different.txt");
let refusal = policy.evaluate_request(&different);
assert!(refusal.is_ask(), "{refusal:?}");
assert_eq!(refusal.receipt["matched_rule"]["id"], "ask");
}
#[test]
fn host_request_unannotated_paths_require_owned_workspace_authority() {
let workspace = tempfile::tempdir().unwrap();
let policy = ToolApprovalPolicy::from_host_json(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [{"id": "user-read", "source": "user", "allow": {"tool": "custom_read"}}]
}))
.unwrap();
let inside = ToolApprovalRequest::from_host_json(serde_json::json!({
"tool_name": "custom_read",
"arguments": {"path": "inside.txt"},
"workspace_boundary": {"root": workspace.path()}
}))
.unwrap();
let allowed = policy.evaluate_request(&inside);
assert!(allowed.is_allow(), "{allowed:?}");
assert_eq!(allowed.receipt["matched_rule"]["id"], "user-read");
let mut external = inside;
external.arguments = serde_json::json!({"path": "../outside.txt"});
let refused = policy.evaluate_request(&external);
assert!(refused.is_deny(), "{refused:?}");
external.workspace_boundary = None;
let no_authority = policy.evaluate_request(&external);
assert!(no_authority.is_deny(), "{no_authority:?}");
assert_eq!(
no_authority.receipt["matched_rule"]["id"],
"invalid_host_request"
);
assert!(ToolApprovalRequest::from_host_json(serde_json::to_value(&external).unwrap()).is_err());
for (arguments, policy_decision) in [
(serde_json::json!({"paths": ["../outside.txt"]}), None),
(serde_json::json!({"path": null}), None),
(serde_json::json!({"path": {}}), None),
(serde_json::json!({"path": ""}), None),
(serde_json::json!({"paths": [123]}), None),
(serde_json::json!({"paths": []}), None),
(serde_json::json!({"command": "cat ../outside.txt"}), None),
(
serde_json::json!({"rawInput": {"path": "../outside.txt"}}),
None,
),
(
serde_json::json!({}),
Some(serde_json::json!({"context": {"path": "../outside.txt"}})),
),
] {
let request = ToolApprovalRequest {
tool_name: "custom_read".into(),
arguments,
policy_decision,
..Default::default()
};
let refused = policy.evaluate_request(&request);
assert!(refused.is_deny(), "{refused:?}");
assert_eq!(
refused.receipt["matched_rule"]["id"],
"invalid_host_request"
);
}
let no_paths = ToolApprovalRequest::from_host_json(serde_json::json!({
"tool_name": "custom_read", "arguments": {"url": "https://example.invalid"}
}))
.unwrap();
assert!(policy.evaluate_request(&no_paths).is_allow());
}
#[test]
fn host_request_path_free_annotations_do_not_require_workspace_authority() {
let policy = ToolApprovalPolicy::from_host_json(serde_json::json!({
"rules": [{"id": "review-command", "ask": {
"tool": "annotated_command", "tool_kind": "execute", "side_effect": "process_exec"
}}]
}))
.unwrap();
let request_json = |arguments: JsonValue, path_params: JsonValue| {
serde_json::json!({
"tool_name": "annotated_command",
"arguments": arguments,
"tool_annotations": {
"kind": "execute", "side_effect_level": "process_exec",
"arg_schema": {"path_params": path_params}
},
"policy_decision": {"context": {"tool_kind": "read", "side_effect": "read_only"}}
})
};
for path_params in [serde_json::json!([]), serde_json::json!(["filename"])] {
let request = ToolApprovalRequest::from_host_json(request_json(
serde_json::json!({"command": "git status"}),
path_params,
))
.expect("annotations without actual paths do not require workspace authority");
let decision = policy.evaluate_request(&request);
assert!(decision.is_ask(), "{decision:?}");
assert_eq!(decision.receipt["matched_rule"]["id"], "review-command");
assert_eq!(decision.receipt["context"]["tool_kind"], "execute");
assert_eq!(decision.receipt["context"]["side_effect"], "process_exec");
}
for value in [
request_json(
serde_json::json!({"filename": "outside.txt"}),
serde_json::json!(["filename"]),
),
serde_json::json!({"tool_name": "annotated_command", "arguments": {"path": "outside.txt"}}),
request_json(
serde_json::json!({"command": "cat ../outside.txt"}),
serde_json::json!([]),
),
] {
let refusal = ToolApprovalRequest::from_host_json(value.clone()).unwrap_err();
assert!(refusal.contains("explicit workspace_boundary"), "{refusal}");
let direct: ToolApprovalRequest = serde_json::from_value(value).unwrap();
let decision = policy.evaluate_request(&direct);
assert!(decision.is_deny(), "{decision:?}");
assert_eq!(
decision.receipt["matched_rule"]["id"],
"invalid_host_request"
);
}
for value in [
serde_json::json!(null),
serde_json::json!({}),
serde_json::json!([1]),
] {
let refusal = ToolApprovalRequest::from_host_json(request_json(
serde_json::json!({"filename": value}),
serde_json::json!(["filename"]),
))
.unwrap_err();
assert!(
refusal.contains("workspace path argument 'filename'"),
"{refusal}"
);
}
let workspace = tempfile::tempdir().unwrap();
let mut sensitive = request_json(
serde_json::json!({"filename": ".env"}),
serde_json::json!(["filename"]),
);
sensitive["workspace_boundary"] = serde_json::json!({"root": workspace.path()});
let sensitive = ToolApprovalRequest::from_host_json(sensitive).unwrap();
let refusal = policy.evaluate_request(&sensitive);
assert!(refusal.is_deny(), "{refusal:?}");
assert!(refusal
.risk_labels
.iter()
.any(|label| label == "sensitive_path"));
}
#[test]
fn host_request_rejects_invalid_explicit_workspace_inputs() {
let workspace = tempfile::tempdir().unwrap();
let valid = serde_json::json!({
"tool_name": "custom_read", "arguments": {"filename": "file.txt"},
"workspace_boundary": {"root": workspace.path().to_str().unwrap()},
"tool_annotations": {"arg_schema": {"path_params": ["filename"]}}
});
assert!(ToolApprovalRequest::from_host_json(valid.clone()).is_ok());
let mut no_root = valid.clone();
no_root
.as_object_mut()
.unwrap()
.remove("workspace_boundary");
assert!(ToolApprovalRequest::from_host_json(no_root).is_err());
let mut malformed_fallback = valid.clone();
malformed_fallback
.as_object_mut()
.unwrap()
.remove("tool_annotations");
malformed_fallback["arguments"] = serde_json::json!({"paths": [123]});
assert!(ToolApprovalRequest::from_host_json(malformed_fallback).is_err());
for root in [
serde_json::json!("relative"),
serde_json::json!(workspace.path().join("absent")),
serde_json::json!("/\0"),
] {
let mut invalid = valid.clone();
invalid["workspace_boundary"]["root"] = root;
assert!(ToolApprovalRequest::from_host_json(invalid).is_err());
}
for value in [
serde_json::json!(null),
serde_json::json!({}),
serde_json::json!(["file.txt", 1]),
serde_json::json!(""),
] {
let mut invalid = valid.clone();
invalid["arguments"]["filename"] = value;
assert!(ToolApprovalRequest::from_host_json(invalid).is_err());
}
let mut unknown = valid.clone();
unknown["tool_annotations"]["arg_schema"]["path_param"] = serde_json::json!(["other"]);
assert!(ToolApprovalRequest::from_host_json(unknown).is_err());
let mut direct = ToolApprovalRequest::from_host_json(valid).unwrap();
direct.workspace_boundary.as_mut().unwrap().root = "relative".into();
let refusal = ToolApprovalPolicy::default().evaluate_request(&direct);
assert!(refusal.is_deny());
assert_eq!(
refusal.receipt["matched_rule"]["id"],
"invalid_host_request"
);
}
#[test]
fn write_environment_allows_require_an_exact_mode_match() {
let wildcard: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [{"id": "wildcard", "allow": {"tool": "run", "env_mode": "*"}}]
}))
.expect("policy");
let exact: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [{"id": "exact", "allow": {"tool": "run", "env_mode": "patch"}}]
}))
.expect("policy");
let request = ToolApprovalRequest {
tool_name: "run".to_string(),
arguments: serde_json::json!({"env_mode": "patch"}),
..Default::default()
};
let wildcard_decision = wildcard.evaluate_request(&request);
let exact_decision = exact.evaluate_request(&request);
assert!(wildcard_decision.is_allow());
assert!(
wildcard_decision.matched_rule.is_none(),
"a wildcard must not grant patch-mode writes"
);
assert_eq!(
exact_decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("exact")
);
}
#[test]
fn approval_request_undo_metadata_is_a_supported_host_boundary() {
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"allow_sensitive_paths": true,
"rules": [{"id": "ask-network", "ask": {"side_effect": "network"}}]
}))
.expect("policy");
let request = ToolApprovalRequest {
tool_name: "fetch".to_string(),
arguments: serde_json::json!({}),
approval_request: Some(serde_json::json!({
"undo_metadata": {
"policy_decision": {
"context": {"side_effect": "network"}
}
}
})),
..Default::default()
};
let decision = policy.evaluate_request(&request);
assert!(decision.is_ask());
assert_eq!(
decision
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some("ask-network")
);
}
#[test]
fn approval_unavailable_class_count_uses_sorted_risk_labels() {
clear_all_approval_policy_repeat_counts();
let labels = vec![
"package_install".to_string(),
"approval_required".to_string(),
];
let reversed = vec![
"approval_required".to_string(),
"package_install".to_string(),
];
let (class, count) = next_approval_unavailable_class_repeat_count("s1", &labels);
assert_eq!(class, "approval_required+package_install");
assert_eq!(count, 1);
let (class, count) = next_approval_unavailable_class_repeat_count("s1", &reversed);
assert_eq!(class, "approval_required+package_install");
assert_eq!(count, 2);
let (class, count) = next_approval_unavailable_class_repeat_count("s2", &[]);
assert_eq!(class, "approval_required");
assert_eq!(count, 1);
clear_approval_policy_repeat_counts("s1");
let (_, count) = next_approval_unavailable_class_repeat_count("s1", &labels);
assert_eq!(count, 1);
}
#[test]
fn approval_repeat_digest_ignores_argument_key_order() {
let first: serde_json::Value = serde_json::from_str(r#"{"zeta":1,"alpha":2}"#).unwrap();
let second: serde_json::Value = serde_json::from_str(r#"{"alpha":2,"zeta":1}"#).unwrap();
assert_eq!(stable_json_digest(&first), stable_json_digest(&second));
}
#[test]
fn each_deciding_source_names_its_own_gate() {
use crate::agent_events::DenialGate;
let temp = tempfile::tempdir().unwrap();
crate::stdlib::process::set_thread_execution_context(Some(
crate::orchestration::RunExecutionRecord {
cwd: Some(temp.path().to_string_lossy().into_owned()),
project_root: None,
source_dir: Some(temp.path().to_string_lossy().into_owned()),
env: BTreeMap::new(),
adapter: None,
repo_path: None,
worktree_path: None,
branch: None,
base_ref: None,
cleanup: None,
environment_policy: Default::default(),
grants: Vec::new(),
admitted_environment: Vec::new(),
},
));
policy_with_path_annotation("read_file", ToolKind::Read);
let external = tempfile::tempdir().unwrap();
let outside_path = external.path().join("outside.txt");
let reported_outside = external
.path()
.canonicalize()
.unwrap()
.join("outside.txt")
.to_string_lossy()
.replace('\\', "/");
let outside = evaluate_tool_approval_policy(
&ToolApprovalPolicy::default(),
"read_file",
&serde_json::json!({"path": outside_path}),
None,
);
assert!(outside.is_deny());
assert_eq!(
outside
.matched_rule
.as_ref()
.map(|rule| rule.source.as_str()),
Some(SOURCE_DEFAULT_EXTERNAL_PATH),
);
assert_eq!(outside.denial_gate(), DenialGate::WorkspaceBoundary);
let sensitive = evaluate_tool_approval_policy(
&ToolApprovalPolicy::default(),
"read_file",
&serde_json::json!({"path": "config/.env"}),
None,
);
assert!(sensitive.is_deny());
assert_eq!(
sensitive
.matched_rule
.as_ref()
.map(|rule| rule.source.as_str()),
Some(SOURCE_DEFAULT_SENSITIVE_PATH),
);
assert_eq!(sensitive.denial_gate(), DenialGate::SensitivePath);
let configured = evaluate_tool_approval_policy(
&ToolApprovalPolicy {
auto_deny: vec!["read_file".to_string()],
allow_sensitive_paths: true,
allow_external_paths: true,
..Default::default()
},
"read_file",
&serde_json::json!({"path": "src/main.rs"}),
None,
);
assert!(configured.is_deny());
assert_eq!(configured.denial_gate(), DenialGate::ApprovalPolicy);
let rendered = [&outside, &sensitive, &configured].map(|decision| {
let denial = decision.terminal_denial();
assert_eq!(denial.gate, decision.denial_gate());
assert!(!denial.retryable, "a policy refusal is terminal");
(denial.reason, denial.denied_paths)
});
assert_eq!(rendered[0].1, vec![reported_outside.clone()]);
assert_eq!(rendered[1].1.len(), 1, "{:?}", rendered[1].1);
assert!(
rendered[1].1[0].ends_with("config/.env"),
"{:?}",
rendered[1].1
);
assert!(rendered[2].1.is_empty(), "{:?}", rendered[2].1);
let rendered = rendered.map(|(reason, _)| reason);
assert!(
rendered[0].starts_with("Workspace boundary denial:"),
"{rendered:?}"
);
assert!(
rendered[1].starts_with("Sensitive path denial:"),
"{rendered:?}"
);
assert!(
rendered[2].starts_with("Approval policy denial:"),
"{rendered:?}"
);
assert!(rendered[0].contains(&reported_outside), "{rendered:?}");
pop_execution_policy();
crate::stdlib::process::set_thread_execution_context(None);
}
#[test]
fn a_ruleless_deny_stays_an_approval_denial() {
use crate::agent_events::DenialGate;
assert_eq!(denial_gate_for_source(None), DenialGate::ApprovalPolicy);
assert_eq!(
denial_gate_for_source(Some("auto_deny")),
DenialGate::ApprovalPolicy,
);
assert_eq!(
denial_gate_for_source(Some("repeat_limit")),
DenialGate::ApprovalPolicy,
);
assert_eq!(
denial_gate_for_source(Some(SOURCE_DEFAULT_PATH_GUARD)),
DenialGate::WorkspaceBoundary,
);
}
#[test]
fn a_path_refusal_names_the_path_that_refused_not_every_path_declared() {
let temp = tempfile::tempdir().unwrap();
crate::stdlib::process::set_thread_execution_context(Some(
crate::orchestration::RunExecutionRecord {
cwd: Some(temp.path().to_string_lossy().into_owned()),
project_root: None,
source_dir: Some(temp.path().to_string_lossy().into_owned()),
env: BTreeMap::new(),
adapter: None,
repo_path: None,
worktree_path: None,
branch: None,
base_ref: None,
cleanup: None,
environment_policy: Default::default(),
grants: Vec::new(),
admitted_environment: Vec::new(),
},
));
let mut annotations = BTreeMap::new();
annotations.insert(
"move_file".to_string(),
ToolAnnotations {
kind: ToolKind::Move,
side_effect_level: SideEffectLevel::WorkspaceWrite,
arg_schema: ToolArgSchema {
path_params: vec!["from".to_string(), "to".to_string()],
..Default::default()
},
..Default::default()
},
);
push_execution_policy(CapabilityPolicy {
tool_annotations: annotations,
..Default::default()
});
let external = tempfile::tempdir().unwrap();
let outside_path = external.path().join("outside.txt");
let reported_outside = external
.path()
.canonicalize()
.unwrap()
.join("outside.txt")
.to_string_lossy()
.replace('\\', "/");
let args = serde_json::json!({"from": "src/main.rs", "to": outside_path});
let decision =
evaluate_tool_approval_policy(&ToolApprovalPolicy::default(), "move_file", &args, None);
assert!(decision.is_deny());
let declared = crate::orchestration::current_tool_declared_paths("move_file", &args);
assert!(
declared.len() > 1,
"this test measures nothing unless the call declares more than one path: {declared:?}"
);
let denial = decision.terminal_denial();
assert_eq!(
denial.denied_paths,
vec![reported_outside],
"the refusal must name the path that refused, not the declared set {declared:?}"
);
pop_execution_policy();
crate::stdlib::process::set_thread_execution_context(None);
}
fn with_workspace_and_external_root<T>(run: impl FnOnce(&str, &str) -> T) -> T {
let workspace = tempfile::tempdir().unwrap();
let external = tempfile::tempdir().unwrap();
crate::stdlib::process::set_thread_execution_context(Some(
crate::orchestration::RunExecutionRecord {
cwd: Some(workspace.path().to_string_lossy().into_owned()),
source_dir: Some(workspace.path().to_string_lossy().into_owned()),
..Default::default()
},
));
let external_root = external
.path()
.canonicalize()
.unwrap()
.to_string_lossy()
.into_owned();
let inside = external
.path()
.canonicalize()
.unwrap()
.join("notes.txt")
.to_string_lossy()
.into_owned();
let result = run(&external_root, &inside);
crate::stdlib::process::set_thread_execution_context(None);
result
}
fn evaluate_annotated(
policy: &ToolApprovalPolicy,
tool: &str,
kind: ToolKind,
path: &str,
) -> PolicyEvaluation {
policy_with_path_annotation(tool, kind);
let decision =
evaluate_tool_approval_policy(policy, tool, &serde_json::json!({ "path": path }), None);
pop_execution_policy();
decision
}
#[test]
fn a_string_external_root_admits_reads_and_refuses_writes_at_the_boundary() {
with_workspace_and_external_root(|root, inside| {
let policy: ToolApprovalPolicy =
serde_json::from_value(serde_json::json!({ "external_roots": [root] })).unwrap();
assert!(policy.rules.is_empty() && policy.auto_deny.is_empty());
let read = evaluate_annotated(&policy, "read_file", ToolKind::Read, inside);
assert!(read.is_allow(), "read under a read root: {}", read.reason);
assert_eq!(
read.receipt["context"]["external_roots"],
serde_json::json!([{ "path": root, "access": "read" }])
);
let write = evaluate_annotated(&policy, "write_file", ToolKind::Edit, inside);
assert!(write.is_deny(), "write under a read root: {}", write.reason);
let rule = write.matched_rule.as_ref().expect("the boundary decided");
assert_eq!(rule.source, SOURCE_DEFAULT_EXTERNAL_PATH);
assert_eq!(rule.id.as_deref(), Some(EXTERNAL_ROOT_READ_ONLY));
assert_eq!(write.risk_labels, vec![EXTERNAL_ROOT_READ_ONLY.to_string()]);
let reported = inside.replace('\\', "/");
assert_eq!(write.denied_paths, vec![reported.clone()]);
assert!(
write.reason.contains(&reported)
&& write.reason.contains(root)
&& write.reason.contains("'read'"),
"the refusal names the path, the root, and its mode: {}",
write.reason
);
assert_eq!(
write.denial_gate(),
crate::agent_events::DenialGate::WorkspaceBoundary
);
assert_eq!(
write.receipt["context"]["external_roots"],
serde_json::json!([{ "path": root, "access": "read" }])
);
});
}
#[test]
fn a_read_write_external_root_admits_writes() {
with_workspace_and_external_root(|root, inside| {
let policy: ToolApprovalPolicy = serde_json::from_value(serde_json::json!({
"external_roots": [{ "path": root, "access": "read_write" }]
}))
.unwrap();
let write = evaluate_annotated(&policy, "write_file", ToolKind::Edit, inside);
assert!(
write.is_allow(),
"write under a read_write root: {}",
write.reason
);
assert_eq!(
write.receipt["context"]["external_roots"],
serde_json::json!([{ "path": root, "access": "read_write" }])
);
});
}
#[test]
fn a_read_root_still_refuses_writes_when_external_paths_are_opted_out() {
with_workspace_and_external_root(|root, inside| {
let policy = ToolApprovalPolicy {
allow_external_paths: true,
external_roots: vec![ExternalRoot::read(root)],
..Default::default()
};
let write = evaluate_annotated(&policy, "write_file", ToolKind::Edit, inside);
assert!(write.is_deny(), "{}", write.reason);
assert_eq!(
write
.matched_rule
.as_ref()
.and_then(|rule| rule.id.as_deref()),
Some(EXTERNAL_ROOT_READ_ONLY)
);
});
}
#[test]
fn intersecting_policies_keeps_the_narrower_mode_for_a_shared_root() {
let host = ToolApprovalPolicy {
external_roots: vec![ExternalRoot::read("/opt/shared")],
..Default::default()
};
let session = ToolApprovalPolicy {
external_roots: vec![ExternalRoot::read_write("/opt/shared")],
..Default::default()
};
assert_eq!(
host.intersect(&session).external_roots,
vec![ExternalRoot::read("/opt/shared")]
);
assert_eq!(
session.intersect(&host).external_roots,
vec![ExternalRoot::read("/opt/shared")]
);
}