#[cfg(target_os = "linux")]
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use crate::permissions::{ApprovalOutcome, ApprovalRequest, PermissionsApprovalHandler};
use crate::tools::SandboxPolicy;
#[derive(Clone)]
pub struct SandboxApprovalHandler(pub Arc<dyn PermissionsApprovalHandler>);
impl std::fmt::Debug for SandboxApprovalHandler {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("SandboxApprovalHandler(..)")
}
}
impl std::ops::Deref for SandboxApprovalHandler {
type Target = dyn PermissionsApprovalHandler;
fn deref(&self) -> &Self::Target {
&*self.0
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SandboxEscalation {
#[default]
Deny,
Ask,
Allow,
}
impl SandboxEscalation {
pub fn rank(self) -> u8 {
match self {
SandboxEscalation::Deny => 0,
SandboxEscalation::Ask => 1,
SandboxEscalation::Allow => 2,
}
}
pub fn parse(s: &str) -> Option<Self> {
match s.replace('_', "-").to_ascii_lowercase().as_str() {
"deny" => Some(SandboxEscalation::Deny),
"ask" => Some(SandboxEscalation::Ask),
"allow" => Some(SandboxEscalation::Allow),
_ => None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SandboxEnvPolicy {
#[default]
Inherit,
Filtered,
None,
}
impl SandboxEnvPolicy {
pub fn rank(self) -> u8 {
match self {
SandboxEnvPolicy::None => 0,
SandboxEnvPolicy::Filtered => 1,
SandboxEnvPolicy::Inherit => 2,
}
}
pub fn parse(s: &str) -> Option<Self> {
match s.replace('_', "-").to_ascii_lowercase().as_str() {
"inherit" => Some(SandboxEnvPolicy::Inherit),
"filtered" => Some(SandboxEnvPolicy::Filtered),
"none" => Some(SandboxEnvPolicy::None),
_ => None,
}
}
}
const FILTERED_ENV_DENYLIST_PREFIXES: &[&str] = &[
"OPENROUTER_API_KEY",
"OPENAI_API_KEY",
"ANTHROPIC_API_KEY",
"AWS_ACCESS_KEY_ID",
"AWS_SECRET_ACCESS_KEY",
"AWS_SESSION_TOKEN",
"GITHUB_TOKEN",
"GH_TOKEN",
"GITLAB_TOKEN",
"NPM_TOKEN",
"DOCKER_PASSWORD",
"GOOGLE_APPLICATION_CREDENTIALS",
"AZURE_CLIENT_SECRET",
"SSH_AUTH_SOCK",
"SUPERCODE_",
];
fn is_filtered_env_key(key: &str) -> bool {
let upper = key.to_ascii_uppercase();
if FILTERED_ENV_DENYLIST_PREFIXES
.iter()
.any(|p| upper == *p || upper.starts_with(p))
{
return true;
}
[
"TOKEN",
"SECRET",
"PASSWORD",
"_KEY",
"CREDENTIAL",
"APIKEY",
]
.iter()
.any(|needle| upper.contains(needle))
}
const MINIMAL_ENV_KEEP: &[&str] = &["PATH", "HOME", "TERM", "LANG", "LC_ALL", "TMPDIR"];
pub fn apply_env_policy<I, K, V>(policy: SandboxEnvPolicy, base: I) -> Vec<(String, String)>
where
I: IntoIterator<Item = (K, V)>,
K: Into<String>,
V: Into<String>,
{
let base: Vec<(String, String)> = base
.into_iter()
.map(|(k, v)| (k.into(), v.into()))
.collect();
match policy {
SandboxEnvPolicy::Inherit => base,
SandboxEnvPolicy::Filtered => base
.into_iter()
.filter(|(k, _)| !is_filtered_env_key(k))
.collect(),
SandboxEnvPolicy::None => base
.into_iter()
.filter(|(k, _)| MINIMAL_ENV_KEEP.contains(&k.as_str()))
.collect(),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FsDecision {
NotRequested,
Confine,
RunUnconfinedWithWarning {
reason: String,
},
Refuse {
reason: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum NetDecision {
NotRequested,
Confine,
GapWarn {
reason: String,
},
}
pub fn os_sandbox_active(tier: SandboxPolicy, os_enabled: Option<bool>) -> bool {
match tier {
SandboxPolicy::DangerFullAccess => false,
_ => os_enabled.unwrap_or(true),
}
}
#[allow(clippy::too_many_arguments)]
pub fn decide_fs(
tier: SandboxPolicy,
os_enabled: Option<bool>,
fs_available: bool,
escalation: SandboxEscalation,
approval: Option<&dyn PermissionsApprovalHandler>,
subject: &str,
) -> FsDecision {
if !os_sandbox_active(tier, os_enabled) {
return FsDecision::NotRequested;
}
if fs_available {
return FsDecision::Confine;
}
let reason = format!(
"sandbox: filesystem confinement ({tier:?}) was requested but is unavailable on this \
platform/kernel (no Landlock support) for `{subject}`"
);
resolve_escalation(escalation, approval, "bash", subject, reason)
}
pub fn decide_net(
network_enabled: bool,
has_domain_rules: bool,
net_available: bool,
) -> NetDecision {
if !network_enabled {
return NetDecision::NotRequested;
}
if has_domain_rules {
return NetDecision::GapWarn {
reason: "sandbox: capabilities.permissions.sandbox.network.allow_domains/\
deny_domains was set, but domain-level network filtering has no OS \
primitive on this platform — that needs an out-of-scope TLS-MITM proxy \
(COMPOSABLE-HARNESS-DESIGN.md gap honesty note). Network was NOT \
confined for this call."
.to_string(),
};
}
if net_available {
return NetDecision::Confine;
}
NetDecision::GapWarn {
reason: "sandbox: capabilities.permissions.sandbox.network.enabled was set, but a \
coarse network cut-off is unavailable on this platform/kernel (no \
unprivileged network-namespace support). Network was NOT confined for this \
call."
.to_string(),
}
}
fn resolve_escalation(
escalation: SandboxEscalation,
approval: Option<&dyn PermissionsApprovalHandler>,
tool: &str,
subject: &str,
reason: String,
) -> FsDecision {
match escalation {
SandboxEscalation::Deny => FsDecision::Refuse { reason },
SandboxEscalation::Allow => FsDecision::RunUnconfinedWithWarning { reason },
SandboxEscalation::Ask => match approval {
Some(handler) => {
let raw_args = serde_json::Value::Null;
let req = ApprovalRequest {
tool,
subject: Some(subject),
raw_args: &raw_args,
};
match handler.ask(&req) {
ApprovalOutcome::Deny => FsDecision::Refuse { reason },
ApprovalOutcome::Allow | ApprovalOutcome::AllowForSession => {
FsDecision::RunUnconfinedWithWarning { reason }
}
}
}
None => FsDecision::Refuse { reason },
},
}
}
pub fn warn_once(reason: &str) {
static WARNED: OnceLock<Mutex<std::collections::HashSet<String>>> = OnceLock::new();
let set = WARNED.get_or_init(|| Mutex::new(std::collections::HashSet::new()));
if let Ok(mut set) = set.lock() {
if set.insert(reason.to_string()) {
eprintln!("\x1b[33mwarning: {reason}\x1b[0m");
}
}
}
#[cfg(target_os = "linux")]
pub fn landlock_available() -> bool {
static AVAILABLE: OnceLock<bool> = OnceLock::new();
*AVAILABLE.get_or_init(|| {
use landlock::{AccessFs, CompatLevel, Compatible, Ruleset, RulesetAttr, ABI};
Ruleset::default()
.set_compatibility(CompatLevel::HardRequirement)
.handle_access(AccessFs::from_write(ABI::V1))
.and_then(|r| r.create())
.is_ok()
})
}
#[cfg(not(target_os = "linux"))]
pub fn landlock_available() -> bool {
false
}
#[cfg(target_os = "linux")]
pub fn netns_available() -> bool {
static AVAILABLE: OnceLock<bool> = OnceLock::new();
*AVAILABLE.get_or_init(probe_netns_fork)
}
#[cfg(not(target_os = "linux"))]
pub fn netns_available() -> bool {
false
}
#[cfg(target_os = "linux")]
fn probe_netns_fork() -> bool {
unsafe {
let pid = libc::fork();
if pid == 0 {
let rc = libc::unshare(libc::CLONE_NEWUSER | libc::CLONE_NEWNET);
libc::_exit(i32::from(rc != 0));
} else if pid > 0 {
let mut status: libc::c_int = 0;
if libc::waitpid(pid, &mut status, 0) != pid {
return false;
}
libc::WIFEXITED(status) && libc::WEXITSTATUS(status) == 0
} else {
false
}
}
}
#[cfg(target_os = "linux")]
pub fn apply_linux_confinement(
cmd: &mut tokio::process::Command,
confine_fs: bool,
fs_allow_writes: bool,
cwd: PathBuf,
extra_write_dirs: Vec<PathBuf>,
confine_net: bool,
) {
if !confine_fs && !confine_net {
return;
}
let uid = unsafe { libc::getuid() };
let gid = unsafe { libc::getgid() };
unsafe {
cmd.pre_exec(move || {
if confine_net {
netns_isolate_self(uid, gid)
.map_err(|e| std::io::Error::other(format!("sandbox netns: {e}")))?;
}
if confine_fs {
landlock_restrict_self(&cwd, &extra_write_dirs, fs_allow_writes)
.map_err(|e| std::io::Error::other(format!("sandbox landlock: {e}")))?;
}
Ok(())
});
}
}
#[cfg(target_os = "linux")]
fn netns_isolate_self(uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {
unsafe {
if libc::unshare(libc::CLONE_NEWUSER | libc::CLONE_NEWNET) != 0 {
return Err(format!(
"unshare(CLONE_NEWUSER|CLONE_NEWNET): errno {}",
*libc::__errno_location()
));
}
}
write_proc_self_raw("setgroups", b"deny")?;
write_proc_self_raw("uid_map", format!("0 {uid} 1\n").as_bytes())?;
write_proc_self_raw("gid_map", format!("0 {gid} 1\n").as_bytes())?;
Ok(())
}
#[cfg(target_os = "linux")]
fn write_proc_self_raw(name: &str, contents: &[u8]) -> Result<(), String> {
let path = format!("/proc/self/{name}\0");
unsafe {
let fd = libc::open(path.as_ptr() as *const libc::c_char, libc::O_WRONLY);
if fd < 0 {
return Err(format!(
"open(/proc/self/{name}): errno {}",
*libc::__errno_location()
));
}
let n = libc::write(fd, contents.as_ptr() as *const libc::c_void, contents.len());
let write_errno = *libc::__errno_location();
libc::close(fd);
if n != contents.len() as isize {
return Err(format!("write(/proc/self/{name}): errno {write_errno}"));
}
}
Ok(())
}
#[cfg(target_os = "linux")]
fn landlock_restrict_self(
cwd: &Path,
extra_write_dirs: &[PathBuf],
fs_allow_writes: bool,
) -> Result<(), String> {
use landlock::{
path_beneath_rules, AccessFs, CompatLevel, Compatible, Ruleset, RulesetAttr,
RulesetCreatedAttr, RulesetStatus, ABI,
};
let write_access = AccessFs::from_write(ABI::V1);
let created = Ruleset::default()
.set_compatibility(CompatLevel::HardRequirement)
.handle_access(write_access)
.map_err(|e| e.to_string())?
.create()
.map_err(|e| e.to_string())?
.set_compatibility(CompatLevel::HardRequirement);
let created = if fs_allow_writes {
let mut dirs = Vec::with_capacity(1 + extra_write_dirs.len());
dirs.push(cwd.to_path_buf());
dirs.extend(extra_write_dirs.iter().cloned());
created
.add_rules(path_beneath_rules(&dirs, write_access))
.map_err(|e| e.to_string())?
} else {
created
};
let status = created.restrict_self().map_err(|e| e.to_string())?;
if status.ruleset != RulesetStatus::FullyEnforced {
return Err(format!(
"ruleset not fully enforced ({:?}) — refusing to claim confinement it doesn't have",
status.ruleset
));
}
Ok(())
}