use crate::{Caveats, SandboxPolicy, ToolResult};
use std::sync::Arc;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SandboxKind {
Landlock,
Seatbelt,
AppContainer,
MinimalRootfs,
MicroVm,
#[default]
None,
}
pub trait Sandbox: Send + Sync {
fn kind(&self) -> SandboxKind;
fn apply(&self, effective: &Caveats) -> ToolResult<()>;
fn command_prefix(&self, effective: &Caveats) -> ToolResult<Vec<String>> {
let _ = effective;
Ok(Vec::new())
}
}
#[derive(Debug, Default, Clone, Copy)]
pub struct NoopSandbox;
impl Sandbox for NoopSandbox {
fn kind(&self) -> SandboxKind {
SandboxKind::None
}
fn apply(&self, _effective: &Caveats) -> ToolResult<()> {
Ok(())
}
}
#[must_use]
pub(crate) fn restricts_fs(caveats: &Caveats) -> bool {
matches!(caveats.fs_write, crate::Scope::Only(_))
|| matches!(caveats.fs_read, crate::Scope::Only(_))
}
#[must_use]
pub(crate) fn restricts_exec(caveats: &Caveats) -> bool {
matches!(caveats.exec, crate::Scope::Only(_))
}
#[must_use]
pub(crate) fn net_fully_denied(caveats: &Caveats) -> bool {
matches!(&caveats.net, crate::Scope::Only(s) if s.is_empty())
}
#[must_use]
pub(crate) fn exec_fully_denied(caveats: &Caveats) -> bool {
matches!(&caveats.exec, crate::Scope::Only(s) if s.is_empty())
}
#[cfg(all(target_os = "linux", feature = "linux-landlock"))]
pub(crate) fn landlock_net_capable() -> bool {
landlock_impl::landlock_net_is_supported()
}
#[cfg(not(all(target_os = "linux", feature = "linux-landlock")))]
pub(crate) fn landlock_net_capable() -> bool {
false
}
pub(crate) const LOOPBACK_HOSTS: &[&str] = &["localhost", "127.0.0.1", "::1"];
#[must_use]
pub(crate) fn net_loopback_only(caveats: &Caveats) -> bool {
matches!(&caveats.net, crate::Scope::Only(s)
if !s.is_empty() && s.iter().all(|h| LOOPBACK_HOSTS.contains(&h.as_str())))
}
#[must_use]
pub fn net_egress_proxy_hosts(caveats: &Caveats) -> Option<Vec<String>> {
match &caveats.net {
crate::Scope::Only(s)
if !s.is_empty() && s.iter().any(|h| !LOOPBACK_HOSTS.contains(&h.as_str())) =>
{
Some(s.iter().cloned().collect())
}
_ => None,
}
}
#[must_use]
pub fn loopback_fenced_caveats(caveats: &Caveats) -> Caveats {
Caveats {
net: crate::Scope::Only(LOOPBACK_HOSTS.iter().map(|h| (*h).to_string()).collect()),
..caveats.clone()
}
}
#[must_use]
pub fn effective_sandbox_kind(available: SandboxKind, caveats: &Caveats) -> SandboxKind {
match available {
SandboxKind::Landlock
if restricts_fs(caveats) || (net_fully_denied(caveats) && landlock_net_capable()) =>
{
SandboxKind::Landlock
}
SandboxKind::Seatbelt
if restricts_fs(caveats)
|| net_fully_denied(caveats)
|| net_loopback_only(caveats)
|| restricts_exec(caveats) =>
{
SandboxKind::Seatbelt
}
SandboxKind::AppContainer
if net_fully_denied(caveats)
|| net_loopback_only(caveats)
|| exec_fully_denied(caveats)
|| restricts_fs(caveats) =>
{
SandboxKind::AppContainer
}
_ => SandboxKind::None,
}
}
pub fn best_available_sandbox(policy: &Arc<SandboxPolicy>) -> Box<dyn Sandbox> {
#[cfg(all(target_os = "windows", feature = "windows-appcontainer"))]
{
let _ = policy; Box::new(appcontainer_impl::AppContainerSandbox::new())
}
#[cfg(not(all(target_os = "windows", feature = "windows-appcontainer")))]
{
#[cfg(all(target_os = "linux", feature = "linux-landlock"))]
{
if landlock_impl::landlock_is_supported() {
return Box::new(landlock_impl::LandlockSandbox::with_policy(policy.clone()));
}
}
#[cfg(all(target_os = "macos", feature = "macos-seatbelt"))]
{
if seatbelt_impl::seatbelt_is_supported() {
return Box::new(seatbelt_impl::SeatbeltSandbox::with_policy(policy.clone()));
}
}
let _ = policy; Box::new(NoopSandbox)
}
}
#[cfg(all(target_os = "linux", feature = "linux-landlock"))]
pub use landlock_impl::{landlock_is_supported, landlock_net_is_supported, LandlockSandbox};
#[cfg(all(target_os = "macos", feature = "macos-seatbelt"))]
pub use seatbelt_impl::{seatbelt_is_supported, SeatbeltSandbox};
#[cfg(all(target_os = "windows", feature = "windows-appcontainer"))]
pub(crate) mod appcontainer_impl {
use std::sync::atomic::{AtomicU64, Ordering};
use super::{
exec_fully_denied, net_fully_denied, net_loopback_only, restricts_fs, Sandbox, SandboxKind,
};
use crate::{Caveats, Scope, ToolError, ToolResult};
static SPAWN_N: AtomicU64 = AtomicU64::new(0);
#[derive(Debug, Default, Clone, Copy)]
pub struct AppContainerSandbox;
impl AppContainerSandbox {
pub fn new() -> Self {
Self
}
}
fn find_launcher() -> Option<String> {
const LAUNCHER: &str = "agent-bridle-aclaunch.exe";
if let Ok(mut p) = std::env::current_exe() {
p.set_file_name(LAUNCHER);
if p.exists() {
return Some(p.to_string_lossy().into_owned());
}
}
std::env::split_paths(&std::env::var_os("PATH").unwrap_or_default())
.map(|dir| dir.join(LAUNCHER))
.find(|p| p.exists())
.map(|p| p.to_string_lossy().into_owned())
}
impl Sandbox for AppContainerSandbox {
fn kind(&self) -> SandboxKind {
SandboxKind::AppContainer
}
fn apply(&self, _effective: &Caveats) -> ToolResult<()> {
Ok(())
}
fn command_prefix(&self, effective: &Caveats) -> ToolResult<Vec<String>> {
if !net_fully_denied(effective)
&& !net_loopback_only(effective)
&& !exec_fully_denied(effective)
&& !restricts_fs(effective)
{
return Ok(Vec::new());
}
let launcher = find_launcher().ok_or_else(|| {
ToolError::denied(
"windows-appcontainer: agent-bridle-aclaunch.exe not found next to the \
current executable or on PATH; cannot confine",
)
})?;
let n = SPAWN_N.fetch_add(1, Ordering::Relaxed);
let container_name = format!("ab{}{}", std::process::id(), n);
let mut prefix = vec![launcher, "--name".to_string(), container_name];
if matches!(effective.net, Scope::All) {
prefix.push("--net-allow".to_string());
}
if net_loopback_only(effective) {
prefix.push("--loopback-exemption".to_string());
}
if exec_fully_denied(effective) {
prefix.push("--no-child-process".to_string());
}
if let Scope::Only(paths) = &effective.fs_write {
for p in paths {
prefix.push("--fs-write".to_string());
prefix.push(p.clone());
}
}
let write_set: std::collections::HashSet<&str> =
if let Scope::Only(paths) = &effective.fs_write {
paths.iter().map(String::as_str).collect()
} else {
std::collections::HashSet::new()
};
if let Scope::Only(paths) = &effective.fs_read {
for p in paths {
if !write_set.contains(p.as_str()) {
prefix.push("--fs-read".to_string());
prefix.push(p.clone());
}
}
}
Ok(prefix)
}
}
}
#[cfg(all(target_os = "linux", feature = "linux-landlock"))]
pub(crate) mod landlock_impl {
use super::{Sandbox, SandboxKind};
use crate::{Caveats, SandboxPolicy, Scope, ToolError, ToolResult};
use landlock::{
path_beneath_rules, Access, AccessFs, AccessNet, CompatLevel, Compatible, Ruleset,
RulesetAttr, RulesetCreatedAttr, RulesetStatus, ABI,
};
use std::sync::Arc;
fn abi_from_u32(v: u32) -> ABI {
match v {
0 | 1 => ABI::V1,
2 => ABI::V2,
3 => ABI::V3,
4 => ABI::V4,
5 => ABI::V5,
6 => ABI::V6,
_ => ABI::V7,
}
}
fn fs_abi_floor(policy: &SandboxPolicy) -> ABI {
abi_from_u32(policy.landlock_abi_floor.max(3))
}
fn net_abi_floor(policy: &SandboxPolicy) -> ABI {
abi_from_u32(policy.landlock_net_abi_floor.max(4))
}
pub fn landlock_is_supported() -> bool {
Ruleset::default()
.set_compatibility(CompatLevel::HardRequirement)
.handle_access(AccessFs::from_all(ABI::V1))
.and_then(|r| r.create())
.is_ok()
}
pub fn landlock_net_is_supported() -> bool {
Ruleset::default()
.set_compatibility(CompatLevel::HardRequirement)
.handle_access(AccessNet::from_all(ABI::V4))
.and_then(|r| r.create())
.is_ok()
}
#[derive(Debug, Default, Clone)]
pub struct LandlockSandbox {
policy: Arc<SandboxPolicy>,
}
impl LandlockSandbox {
pub fn new() -> Self {
Self::default()
}
pub fn with_policy(policy: Arc<SandboxPolicy>) -> Self {
Self { policy }
}
}
impl Sandbox for LandlockSandbox {
fn kind(&self) -> SandboxKind {
SandboxKind::Landlock
}
fn apply(&self, effective: &Caveats) -> ToolResult<()> {
let write = AccessFs::from_write(fs_abi_floor(&self.policy));
let read = AccessFs::ReadFile | AccessFs::ReadDir;
let confine_read = matches!(effective.fs_read, Scope::Only(_));
let confine_exec = matches!(effective.exec, Scope::Only(_));
let confine_net = super::net_fully_denied(effective);
let mut handled = write;
if confine_read {
handled |= read;
}
if confine_exec {
handled |= AccessFs::Execute;
}
let write_roots = scope_roots(&effective.fs_write);
let ruleset = Ruleset::default()
.set_compatibility(CompatLevel::BestEffort)
.handle_access(handled)
.map_err(landlock_denied)?;
let ruleset = if confine_net {
ruleset
.handle_access(AccessNet::from_all(net_abi_floor(&self.policy)))
.map_err(landlock_denied)?
} else {
ruleset
};
let ruleset = ruleset
.create()
.map_err(landlock_denied)?
.add_rules(path_beneath_rules(&write_roots, write))
.map_err(landlock_denied)?;
let ruleset = if confine_read {
let mut read_roots = scope_roots(&effective.fs_read);
read_roots.extend(self.policy.base_read_paths.resolve());
if confine_exec {
read_roots.extend(resolve_exec_paths(&effective.exec));
} else {
read_roots.extend(self.policy.bin_read_paths.resolve());
}
read_roots.retain(|p| std::path::Path::new(p).exists());
ruleset
.add_rules(path_beneath_rules(&read_roots, read))
.map_err(landlock_denied)?
} else {
ruleset
};
let ruleset = if confine_exec {
let mut exec_roots = resolve_exec_paths(&effective.exec);
exec_roots.extend(self.policy.loader_paths.resolve());
exec_roots.retain(|p| std::path::Path::new(p).exists());
ruleset
.add_rules(path_beneath_rules(&exec_roots, AccessFs::Execute))
.map_err(landlock_denied)?
} else {
ruleset
};
let status = ruleset.restrict_self().map_err(landlock_denied)?;
if status.ruleset == RulesetStatus::NotEnforced {
return Err(ToolError::denied(
"landlock ruleset was not enforced by this kernel",
));
}
Ok(())
}
}
fn resolve_exec_paths(scope: &Scope<String>) -> Vec<String> {
let set = match scope {
Scope::All => return Vec::new(),
Scope::Only(set) => set,
};
let dirs = exec_search_dirs();
let mut out = Vec::new();
for entry in set {
let candidate = if entry.contains('/') {
let p = std::path::PathBuf::from(entry);
p.exists().then_some(p)
} else {
dirs.iter()
.map(|d| std::path::Path::new(d).join(entry))
.find(|c| c.is_file())
};
if let Some(p) = candidate {
if let Ok(canon) = p.canonicalize() {
out.push(canon.to_string_lossy().into_owned());
}
}
}
out
}
fn exec_search_dirs() -> Vec<String> {
if let Ok(path) = std::env::var("PATH") {
let dirs: Vec<String> = path
.split(':')
.filter(|s| !s.is_empty())
.map(String::from)
.collect();
if !dirs.is_empty() {
return dirs;
}
}
[
"/usr/local/bin",
"/usr/bin",
"/bin",
"/usr/local/sbin",
"/usr/sbin",
"/sbin",
]
.iter()
.map(|s| (*s).to_string())
.collect()
}
fn scope_roots(scope: &Scope<String>) -> Vec<String> {
match scope {
Scope::All => vec!["/".to_string()],
Scope::Only(set) => set
.iter()
.filter(|p| std::path::Path::new(p).exists())
.cloned()
.collect(),
}
}
fn landlock_denied(e: impl std::fmt::Display) -> ToolError {
ToolError::denied(format!("landlock: {e}"))
}
}
#[cfg(all(target_os = "macos", feature = "macos-seatbelt"))]
mod seatbelt_impl {
use super::{Sandbox, SandboxKind};
use crate::{Caveats, SandboxPolicy, Scope, ToolError, ToolResult};
use std::path::Path;
use std::sync::Arc;
const SANDBOX_EXEC: &str = "/usr/bin/sandbox-exec";
#[must_use]
pub fn seatbelt_is_supported() -> bool {
Path::new(SANDBOX_EXEC).exists()
}
#[derive(Debug, Default, Clone)]
pub struct SeatbeltSandbox {
policy: Arc<SandboxPolicy>,
}
impl SeatbeltSandbox {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn with_policy(policy: Arc<SandboxPolicy>) -> Self {
Self { policy }
}
}
impl Sandbox for SeatbeltSandbox {
fn kind(&self) -> SandboxKind {
SandboxKind::Seatbelt
}
fn apply(&self, _effective: &Caveats) -> ToolResult<()> {
Ok(())
}
fn command_prefix(&self, effective: &Caveats) -> ToolResult<Vec<String>> {
if !super::restricts_fs(effective)
&& !super::net_fully_denied(effective)
&& !super::net_loopback_only(effective)
&& !super::restricts_exec(effective)
{
return Ok(Vec::new());
}
if !seatbelt_is_supported() {
return Err(ToolError::denied(
"macOS seatbelt: /usr/bin/sandbox-exec is unavailable; cannot confine",
));
}
Ok(vec![
SANDBOX_EXEC.to_string(),
"-p".to_string(),
seatbelt_profile_with(effective, &self.policy.base_read_paths.resolve()),
])
}
}
#[cfg(test)]
#[must_use]
pub fn seatbelt_profile(effective: &Caveats) -> String {
seatbelt_profile_with(
effective,
&SandboxPolicy::default().base_read_paths.resolve(),
)
}
#[must_use]
fn seatbelt_profile_with(effective: &Caveats, base_read: &[String]) -> String {
let mut p = String::from("(version 1)\n(allow default)\n");
if let Scope::Only(_) = &effective.fs_write {
p.push_str("(deny file-write*)\n");
let roots = confined_roots(&effective.fs_write);
if !roots.is_empty() {
p.push_str("(allow file-write*");
for r in &roots {
p.push_str(&format!(" (subpath {})", sbpl_string(r)));
}
p.push_str(")\n");
}
}
if let Scope::Only(_) = &effective.fs_read {
p.push_str("(deny file-read*)\n");
p.push_str("(allow file-read-metadata)\n");
p.push_str("(allow file-read* (literal \"/\")");
for base in base_read {
if let Some(c) = canonical_path(base) {
p.push_str(&format!(" (subpath {})", sbpl_string(&c)));
}
}
for r in confined_roots(&effective.fs_read) {
p.push_str(&format!(" (subpath {})", sbpl_string(&r)));
}
p.push_str(")\n");
}
if super::net_fully_denied(effective) {
p.push_str("(deny network*)\n");
} else if super::net_loopback_only(effective) {
p.push_str("(deny network*)\n");
p.push_str("(allow network* (remote ip \"localhost:*\"))\n");
}
if let Scope::Only(_) = &effective.exec {
p.push_str("(deny process-exec*)\n");
let targets = resolve_exec_targets(&effective.exec);
if !targets.is_empty() {
p.push_str("(allow process-exec*");
for t in &targets {
p.push_str(&format!(" (literal {})", sbpl_string(t)));
}
p.push_str(")\n");
}
}
p
}
fn confined_roots(scope: &Scope<String>) -> Vec<String> {
let Scope::Only(set) = scope else {
return Vec::new();
};
let mut roots: Vec<String> = set.iter().filter_map(|p| canonical_path(p)).collect();
roots.sort();
roots.dedup();
roots
}
const TRUSTED_EXEC_DIRS: &[&str] = &["/usr/bin", "/bin", "/usr/sbin", "/sbin"];
fn resolve_exec_targets(scope: &Scope<String>) -> Vec<String> {
let Scope::Only(set) = scope else {
return Vec::new();
};
let canon_file = |path: &Path, out: &mut Vec<String>| {
if let Ok(c) = std::fs::canonicalize(path) {
if c.is_file() {
out.push(c.to_string_lossy().into_owned());
}
}
};
let mut out: Vec<String> = Vec::new();
for token in set {
if token.starts_with('/') {
canon_file(Path::new(token), &mut out);
} else if !token.contains('/') {
for dir in TRUSTED_EXEC_DIRS {
canon_file(&Path::new(dir).join(token), &mut out);
}
}
}
out.sort();
out.dedup();
out
}
fn canonical_path(p: &str) -> Option<String> {
let path = Path::new(p);
if let Ok(c) = std::fs::canonicalize(path) {
return Some(c.to_string_lossy().into_owned());
}
let mut tail: Vec<std::ffi::OsString> = Vec::new();
let mut cur = path;
while let Some(parent) = cur.parent() {
if let Some(name) = cur.file_name() {
tail.push(name.to_owned());
}
if let Ok(c) = std::fs::canonicalize(parent) {
let mut resolved = c;
for seg in tail.iter().rev() {
resolved.push(seg);
}
return Some(resolved.to_string_lossy().into_owned());
}
cur = parent;
}
None
}
fn sbpl_string(s: &str) -> String {
let mut out = String::with_capacity(s.len() + 2);
out.push('"');
for ch in s.chars() {
if ch == '\\' || ch == '"' {
out.push('\\');
}
out.push(ch);
}
out.push('"');
out
}
#[cfg(test)]
mod unit {
use super::*;
use crate::Scope;
#[test]
fn unrestricted_caveats_make_no_wrapper() {
assert!(SeatbeltSandbox::new()
.command_prefix(&Caveats::top())
.unwrap()
.is_empty());
}
#[test]
fn command_prefix_widens_the_read_base_from_policy() {
if !seatbelt_is_supported() {
eprintln!("skipping: /usr/bin/sandbox-exec unavailable");
return;
}
let extra = std::env::temp_dir().join("abridle-seatbelt-cfg-widen");
std::fs::create_dir_all(&extra).unwrap();
let extra_str = extra.to_string_lossy().into_owned();
let want = canonical_path(&extra_str).expect("temp dir canonicalizes");
let cav = Caveats {
fs_read: Scope::only(["/usr".to_string()]),
..Caveats::top()
};
let default_prefix = SeatbeltSandbox::new().command_prefix(&cav).unwrap();
assert!(
!default_prefix.iter().any(|a| a.contains(&want)),
"default read base must not include the extra dir: {default_prefix:?}"
);
let mut base = SandboxPolicy::default().base_read_paths;
base.extra.push(extra_str);
let policy = Arc::new(SandboxPolicy {
base_read_paths: base,
..SandboxPolicy::default()
});
let widened_prefix = SeatbeltSandbox::with_policy(policy)
.command_prefix(&cav)
.unwrap();
assert!(
widened_prefix.iter().any(|a| a.contains(&want)),
"config-widened base_read_paths must reach the SBPL profile: {widened_prefix:?}"
);
let _ = std::fs::remove_dir_all(&extra);
}
#[test]
fn empty_net_denies_all_egress_and_engages_the_wrapper() {
let cav = Caveats {
net: Scope::none(),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(prof.contains("(deny network*)"), "{prof}");
assert!(
!SeatbeltSandbox::new()
.command_prefix(&cav)
.unwrap()
.is_empty(),
"net:none must engage the sandbox-exec wrapper"
);
}
#[test]
fn nonempty_net_allowlist_is_not_denied() {
let cav = Caveats {
net: Scope::only(["example.com".to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(
!prof.contains("network"),
"non-loopback net must stay ambient: {prof}"
);
}
#[test]
fn loopback_only_net_confines_to_loopback_and_engages() {
for host in ["localhost", "127.0.0.1", "::1"] {
let cav = Caveats {
net: Scope::only([host.to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(prof.contains("(deny network*)"), "{host}: {prof}");
assert!(
prof.contains("(allow network* (remote ip \"localhost:*\"))"),
"{host}: loopback re-allow missing: {prof}"
);
assert!(
!SeatbeltSandbox::new()
.command_prefix(&cav)
.unwrap()
.is_empty(),
"{host}: a loopback-only net grant must engage the wrapper"
);
}
}
#[test]
fn mixed_loopback_and_remote_host_stays_ambient() {
let cav = Caveats {
net: Scope::only(["localhost".to_string(), "example.com".to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(
!prof.contains("network"),
"a mixed loopback+remote allowlist must stay ambient: {prof}"
);
}
#[test]
fn loopback_fenced_caveats_emit_the_egress_proxy_fence() {
let granted = Caveats {
net: Scope::only(["example.com".to_string()]),
fs_write: Scope::only(["/tmp".to_string()]),
..Caveats::top()
};
assert!(!seatbelt_profile(&granted).contains("network"));
let prof = seatbelt_profile(&super::super::loopback_fenced_caveats(&granted));
assert!(prof.contains("(deny network*)"), "{prof}");
assert!(
prof.contains("(allow network* (remote ip \"localhost:*\"))"),
"fence must re-allow loopback: {prof}"
);
assert!(
prof.contains("(deny file-write*)"),
"fs_write rule must survive the fence: {prof}"
);
}
#[test]
fn restricted_write_yields_sandbox_exec_wrapper() {
let cav = Caveats {
fs_write: Scope::only(["/tmp".to_string()]),
..Caveats::top()
};
let prefix = SeatbeltSandbox::new().command_prefix(&cav).unwrap();
assert_eq!(prefix[0], SANDBOX_EXEC);
assert_eq!(prefix[1], "-p");
assert!(prefix[2].contains("(deny file-write*)"));
}
#[test]
fn profile_denies_then_reallows_write_roots() {
let cav = Caveats {
fs_write: Scope::only(["/tmp".to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(prof.contains("(allow default)"));
assert!(prof.contains("(deny file-write*)"));
assert!(prof.contains("(subpath \"/private/tmp\")"), "{prof}");
assert!(!prof.contains("(deny file-read*)"));
}
#[test]
fn empty_write_scope_denies_all_writes_no_allow() {
let cav = Caveats {
fs_write: Scope::none(),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(prof.contains("(deny file-write*)"));
assert!(
!prof.contains("(allow file-write*"),
"an empty scope must grant no write roots: {prof}"
);
}
#[test]
fn restricted_read_includes_loader_base_and_root_entry() {
let cav = Caveats {
fs_read: Scope::only(["/tmp".to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(prof.contains("(deny file-read*)"));
assert!(prof.contains("(literal \"/\")"), "{prof}");
assert!(prof.contains("(subpath \"/usr\")"), "{prof}");
assert!(prof.contains("(subpath \"/System\")"), "{prof}");
}
#[test]
fn sbpl_string_escapes_quotes_and_backslashes() {
assert_eq!(sbpl_string("/a/b"), "\"/a/b\"");
assert_eq!(sbpl_string("/a\"b"), "\"/a\\\"b\"");
assert_eq!(sbpl_string("/a\\b"), "\"/a\\\\b\"");
}
fn unescaped_quotes(s: &str) -> usize {
let b = s.as_bytes();
(0..b.len())
.filter(|&i| b[i] == b'"' && (i == 0 || b[i - 1] != b'\\'))
.count()
}
#[test]
fn crafted_path_cannot_inject_profile_syntax() {
let cav = Caveats {
fs_write: Scope::only(["/tmp/x\") (allow file-write* (subpath \"/".to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert_eq!(
unescaped_quotes(&prof),
2,
"exactly one structural (subpath \"…\") term — no breakout: {prof}"
);
assert!(
prof.contains("\\\""),
"the crafted quotes must be backslash-escaped: {prof}"
);
}
#[test]
fn restricted_exec_emits_deny_and_allowlist() {
let cav = Caveats {
exec: Scope::only(["/bin/echo".to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(prof.contains("(deny process-exec*)"), "{prof}");
assert!(
prof.contains("(allow process-exec* (literal \"/bin/echo\")"),
"{prof}"
);
}
#[test]
fn bare_name_exec_resolves_through_trusted_dirs() {
let cav = Caveats {
exec: Scope::only(["true".to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(
prof.contains("(literal \"/usr/bin/true\")"),
"bare name must resolve to its trusted-dir absolute path: {prof}"
);
}
#[test]
fn restricted_exec_engages_the_wrapper() {
let cav = Caveats {
exec: Scope::only(["/bin/echo".to_string()]),
..Caveats::top()
};
let prefix = SeatbeltSandbox::new().command_prefix(&cav).unwrap();
assert_eq!(prefix.first().map(String::as_str), Some(SANDBOX_EXEC));
}
#[test]
fn empty_exec_scope_denies_all_exec_with_no_allow() {
let cav = Caveats {
exec: Scope::none(),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(prof.contains("(deny process-exec*)"), "{prof}");
assert!(
!prof.contains("(allow process-exec*"),
"an empty exec scope must grant no exec targets: {prof}"
);
}
#[test]
fn relative_and_unresolvable_exec_grants_are_dropped() {
let cav = Caveats {
exec: Scope::only(["./payload".to_string(), "no-such-binary-xyzzy".to_string()]),
..Caveats::top()
};
let prof = seatbelt_profile(&cav);
assert!(prof.contains("(deny process-exec*)"), "{prof}");
assert!(
!prof.contains("(allow process-exec*"),
"unresolvable/relative grants must not anchor an allow: {prof}"
);
}
#[test]
fn unrestricted_exec_emits_no_exec_rules() {
let prof = seatbelt_profile(&Caveats::top());
assert!(!prof.contains("process-exec"), "{prof}");
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::Scope;
#[test]
fn noop_reports_none_and_never_fails() {
let s = NoopSandbox;
assert_eq!(s.kind(), SandboxKind::None);
assert!(s.apply(&Caveats::top()).is_ok());
}
#[test]
fn net_egress_proxy_hosts_triggers_only_on_a_general_remote_allowlist() {
let with_net = |net| {
net_egress_proxy_hosts(&Caveats {
net,
..Caveats::top()
})
};
assert_eq!(with_net(Scope::All), None);
assert_eq!(with_net(Scope::none()), None); for lo in ["localhost", "127.0.0.1", "::1"] {
assert_eq!(
with_net(Scope::only([lo.to_string()])),
None,
"{lo} is loopback-only"
);
}
assert_eq!(
with_net(Scope::only(["example.com".to_string()])),
Some(vec!["example.com".to_string()])
);
let mixed = with_net(Scope::only([
"example.com".to_string(),
"localhost".to_string(),
]))
.expect("mixed set triggers");
assert_eq!(
mixed.len(),
2,
"the FULL grant is returned, loopback included: {mixed:?}"
);
assert!(
mixed.contains(&"example.com".to_string()) && mixed.contains(&"localhost".to_string())
);
}
#[test]
fn loopback_fenced_caveats_swaps_net_to_loopback_preserving_other_axes() {
let granted = Caveats {
net: Scope::only(["example.com".to_string()]),
fs_write: Scope::only(["/tmp/x".to_string()]),
exec: Scope::only(["git".to_string()]),
..Caveats::top()
};
let fenced = loopback_fenced_caveats(&granted);
assert!(
net_loopback_only(&fenced),
"fenced net must be loopback-only"
);
assert!(
net_egress_proxy_hosts(&fenced).is_none(),
"fenced caveats no longer trigger the proxy"
);
assert_eq!(fenced.fs_write, granted.fs_write);
assert_eq!(fenced.exec, granted.exec);
}
#[test]
fn sandbox_kind_serde_is_snake_case() {
assert_eq!(
serde_json::to_string(&SandboxKind::None).unwrap(),
"\"none\""
);
assert_eq!(
serde_json::to_string(&SandboxKind::Landlock).unwrap(),
"\"landlock\""
);
assert_eq!(
serde_json::to_string(&SandboxKind::Seatbelt).unwrap(),
"\"seatbelt\""
);
assert_eq!(
serde_json::to_string(&SandboxKind::AppContainer).unwrap(),
"\"app_container\""
);
assert_eq!(
serde_json::to_string(&SandboxKind::MinimalRootfs).unwrap(),
"\"minimal_rootfs\""
);
assert_eq!(
serde_json::to_string(&SandboxKind::MicroVm).unwrap(),
"\"micro_vm\""
);
}
#[test]
fn effective_kind_downgrades_to_none_when_no_fs_axis_restricted() {
for available in [
SandboxKind::Landlock,
SandboxKind::Seatbelt,
SandboxKind::AppContainer,
SandboxKind::None,
] {
assert_eq!(
effective_sandbox_kind(available, &Caveats::top()),
SandboxKind::None,
"unrestricted fs must report None for {available:?}"
);
}
let restricted = Caveats {
fs_write: Scope::only(["/w".to_string()]),
..Caveats::top()
};
assert_eq!(
effective_sandbox_kind(SandboxKind::Landlock, &restricted),
SandboxKind::Landlock
);
assert_eq!(
effective_sandbox_kind(SandboxKind::Seatbelt, &restricted),
SandboxKind::Seatbelt
);
assert_eq!(
effective_sandbox_kind(SandboxKind::None, &restricted),
SandboxKind::None
);
let read_only = Caveats {
fs_read: Scope::only(["/r".to_string()]),
..Caveats::top()
};
assert_eq!(
effective_sandbox_kind(SandboxKind::Seatbelt, &read_only),
SandboxKind::Seatbelt
);
let net_denied = Caveats {
net: Scope::none(),
..Caveats::top()
};
assert_eq!(
effective_sandbox_kind(SandboxKind::Seatbelt, &net_denied),
SandboxKind::Seatbelt,
"Seatbelt kernel-denies egress, so net:none engages it"
);
let expected_landlock_net = if landlock_net_capable() {
SandboxKind::Landlock
} else {
SandboxKind::None
};
assert_eq!(
effective_sandbox_kind(SandboxKind::Landlock, &net_denied),
expected_landlock_net,
"Landlock engages for net:none only when V4 TCP-deny support is present"
);
}
#[test]
fn appcontainer_engages_for_loopback_only_net() {
for host in ["localhost", "127.0.0.1", "::1"] {
let loopback_only = Caveats {
net: Scope::only([host.to_string()]),
..Caveats::top()
};
assert_eq!(
effective_sandbox_kind(SandboxKind::AppContainer, &loopback_only),
SandboxKind::AppContainer,
"AppContainer must engage for loopback host {host}"
);
}
let remote = Caveats {
net: Scope::only(["example.com".to_string()]),
..Caveats::top()
};
assert_eq!(
effective_sandbox_kind(SandboxKind::AppContainer, &remote),
SandboxKind::None,
"general remote host must not directly engage AppContainer"
);
}
#[test]
fn loopback_fenced_caveats_engages_appcontainer() {
let remote = Caveats {
net: Scope::only(["example.com".to_string()]),
..Caveats::top()
};
let fenced = loopback_fenced_caveats(&remote);
assert!(
net_loopback_only(&fenced),
"loopback_fenced_caveats must produce a loopback-only net scope"
);
assert_eq!(
effective_sandbox_kind(SandboxKind::AppContainer, &fenced),
SandboxKind::AppContainer,
"loopback-fenced caveats must engage AppContainer"
);
}
#[test]
fn best_available_sandbox_is_a_sandbox() {
let sb = best_available_sandbox(&Arc::new(SandboxPolicy::default()));
assert!(sb.apply(&Caveats::top()).is_ok());
}
#[cfg(all(target_os = "windows", feature = "windows-appcontainer"))]
#[test]
fn windows_appcontainer_feature_selects_appcontainer_backend() {
assert_eq!(
best_available_sandbox(&Arc::new(SandboxPolicy::default())).kind(),
SandboxKind::AppContainer
);
}
}
#[cfg(all(target_os = "linux", feature = "linux-landlock", test))]
mod landlock_kernel_tests {
use super::*;
use crate::Scope;
use std::fs;
use std::path::PathBuf;
fn unique_dir(tag: &str) -> PathBuf {
use std::sync::atomic::{AtomicU64, Ordering};
static N: AtomicU64 = AtomicU64::new(0);
let mut d = std::env::temp_dir();
d.push(format!(
"agent-bridle-ll-{}-{}-{}",
tag,
std::process::id(),
N.fetch_add(1, Ordering::Relaxed)
));
fs::create_dir_all(&d).unwrap();
d
}
#[derive(Debug, PartialEq, Eq)]
enum ProofGate {
Run,
Skip,
Fail,
}
fn proof_gate(supported: bool, required: bool) -> ProofGate {
match (supported, required) {
(true, _) => ProofGate::Run,
(false, true) => ProofGate::Fail,
(false, false) => ProofGate::Skip,
}
}
fn skip_proof_unless_landlock() -> bool {
let required = std::env::var("BRIDLE_REQUIRE_LANDLOCK")
.map(|v| !v.is_empty() && v != "0")
.unwrap_or(false);
match proof_gate(landlock_is_supported(), required) {
ProofGate::Run => false,
ProofGate::Skip => {
eprintln!(
"skipping Landlock proof: kernel lacks Landlock \
(set BRIDLE_REQUIRE_LANDLOCK=1 to require it, as CI does)"
);
true
}
ProofGate::Fail => panic!(
"BRIDLE_REQUIRE_LANDLOCK is set but this kernel lacks Landlock — the \
fs_write/fs_read kernel-enforcement proofs cannot be verified (#74)"
),
}
}
#[test]
fn proof_gate_required_but_unsupported_is_a_failure() {
assert_eq!(proof_gate(true, false), ProofGate::Run);
assert_eq!(proof_gate(true, true), ProofGate::Run);
assert_eq!(proof_gate(false, false), ProofGate::Skip);
assert_eq!(proof_gate(false, true), ProofGate::Fail);
}
#[test]
fn fs_write_is_kernel_enforced_outside_scope_denied_inside_allowed() {
if skip_proof_unless_landlock() {
return;
}
let allowed = unique_dir("allowed");
let forbidden = unique_dir("forbidden");
let allowed_t = allowed.clone();
let forbidden_t = forbidden.clone();
let (inside_ok, outside) = std::thread::spawn(move || {
let cav = Caveats {
fs_write: Scope::only([allowed_t.to_string_lossy().into_owned()]),
..Caveats::top()
};
LandlockSandbox::new().apply(&cav).expect("apply landlock");
let inside = fs::write(allowed_t.join("ok.txt"), b"hi");
let outside = fs::write(forbidden_t.join("escape.txt"), b"nope");
(inside.is_ok(), outside)
})
.join()
.unwrap();
assert!(inside_ok, "writing within fs_write scope must succeed");
let err = outside.expect_err("writing outside fs_write scope must be denied by Landlock");
assert_eq!(
err.kind(),
std::io::ErrorKind::PermissionDenied,
"the denial must come from the kernel (EACCES)"
);
let _ = fs::remove_dir_all(&allowed);
let _ = fs::remove_dir_all(&forbidden);
}
#[test]
fn landlock_config_widens_base_read() {
if skip_proof_unless_landlock() {
return;
}
let allowed = unique_dir("cfg-allowed");
let extra = unique_dir("cfg-extra");
fs::write(extra.join("data.txt"), b"configured").unwrap();
let cav = Caveats {
fs_read: Scope::only([allowed.to_string_lossy().into_owned()]),
..Caveats::top()
};
let (extra_c, cav_c) = (extra.clone(), cav.clone());
let denied = std::thread::spawn(move || {
LandlockSandbox::new().apply(&cav_c).expect("apply");
fs::read(extra_c.join("data.txt"))
})
.join()
.unwrap();
assert!(
denied.is_err(),
"default base read must NOT include the out-of-scope extra dir"
);
let mut base = SandboxPolicy::default().base_read_paths;
base.extra.push(extra.to_string_lossy().into_owned());
let policy = Arc::new(SandboxPolicy {
base_read_paths: base,
..SandboxPolicy::default()
});
let extra_w = extra.clone();
let allowed_read = std::thread::spawn(move || {
LandlockSandbox::with_policy(policy)
.apply(&cav)
.expect("apply");
fs::read(extra_w.join("data.txt"))
})
.join()
.unwrap();
assert!(
allowed_read.is_ok(),
"config-widened base_read_paths must allow the extra dir: {allowed_read:?}"
);
let _ = fs::remove_dir_all(&allowed);
let _ = fs::remove_dir_all(&extra);
}
#[test]
fn empty_fs_write_scope_denies_all_writes() {
if skip_proof_unless_landlock() {
return;
}
let dir = unique_dir("none");
let dir_t = dir.clone();
let outside = std::thread::spawn(move || {
let cav = Caveats {
fs_write: Scope::none(),
..Caveats::top()
};
LandlockSandbox::new().apply(&cav).expect("apply landlock");
fs::write(dir_t.join("x.txt"), b"nope")
})
.join()
.unwrap();
assert_eq!(
outside
.expect_err("empty fs_write must deny all writes")
.kind(),
std::io::ErrorKind::PermissionDenied
);
let _ = fs::remove_dir_all(&dir);
}
#[test]
fn fs_read_is_kernel_enforced_outside_scope_denied_inside_allowed() {
if skip_proof_unless_landlock() {
return;
}
let allowed = unique_dir("read-allowed");
let forbidden = unique_dir("read-forbidden");
fs::write(allowed.join("ok.txt"), b"in-scope").unwrap();
fs::write(forbidden.join("secret.txt"), b"out-of-scope").unwrap();
let allowed_t = allowed.clone();
let forbidden_t = forbidden.clone();
let (inside, outside) = std::thread::spawn(move || {
let cav = Caveats {
fs_read: Scope::only([allowed_t.to_string_lossy().into_owned()]),
..Caveats::top()
};
LandlockSandbox::new().apply(&cav).expect("apply landlock");
let inside = fs::read(allowed_t.join("ok.txt"));
let outside = fs::read(forbidden_t.join("secret.txt"));
(inside, outside)
})
.join()
.unwrap();
assert_eq!(inside.expect("in-scope read must succeed"), b"in-scope");
assert_eq!(
outside
.expect_err("reading outside fs_read scope must be denied by Landlock")
.kind(),
std::io::ErrorKind::PermissionDenied,
"the denial must come from the kernel (EACCES)"
);
let _ = fs::remove_dir_all(&allowed);
let _ = fs::remove_dir_all(&forbidden);
}
#[test]
fn read_confined_binary_still_loads_via_base_allowlist() {
if skip_proof_unless_landlock() {
return;
}
let allowed = unique_dir("rc-allowed");
let forbidden = unique_dir("rc-forbidden");
fs::write(allowed.join("ok.txt"), b"hello\n").unwrap();
fs::write(forbidden.join("secret.txt"), b"nope\n").unwrap();
let allowed_t = allowed.clone();
let forbidden_t = forbidden.clone();
let (inside, outside) = std::thread::spawn(move || {
let cav = Caveats {
fs_read: Scope::only([allowed_t.to_string_lossy().into_owned()]),
..Caveats::top()
};
LandlockSandbox::new().apply(&cav).expect("apply landlock");
let inside = std::process::Command::new("cat")
.arg(allowed_t.join("ok.txt"))
.output();
let outside = std::process::Command::new("cat")
.arg(forbidden_t.join("secret.txt"))
.output();
(inside, outside)
})
.join()
.unwrap();
let inside = inside.expect("cat must still load+run under read confinement");
assert!(
inside.status.success(),
"in-scope cat must succeed: {inside:?}"
);
assert_eq!(inside.stdout, b"hello\n");
let outside = outside.expect("cat launches (loader is allowed) even for a denied target");
assert!(
!outside.status.success(),
"cat of an out-of-scope file must fail (read denied): {outside:?}"
);
let _ = fs::remove_dir_all(&allowed);
let _ = fs::remove_dir_all(&forbidden);
}
#[test]
fn fs_read_all_leaves_reads_ambient() {
if skip_proof_unless_landlock() {
return;
}
let outside_dir = unique_dir("ambient-read");
fs::write(outside_dir.join("readable.txt"), b"still readable").unwrap();
let write_scope = unique_dir("ambient-write");
let outside_t = outside_dir.clone();
let write_t = write_scope.clone();
let read = std::thread::spawn(move || {
let cav = Caveats {
fs_write: Scope::only([write_t.to_string_lossy().into_owned()]),
..Caveats::top() };
LandlockSandbox::new().apply(&cav).expect("apply landlock");
fs::read(outside_t.join("readable.txt"))
})
.join()
.unwrap();
assert_eq!(
read.expect("fs_read: All must leave reads ambient"),
b"still readable"
);
let _ = fs::remove_dir_all(&outside_dir);
let _ = fs::remove_dir_all(&write_scope);
}
#[test]
fn exec_direct_execve_of_ungranted_tool_is_kernel_denied() {
if skip_proof_unless_landlock() {
return;
}
let dir = unique_dir("exec");
fs::write(dir.join("data.txt"), b"payload\n").unwrap();
let dir_t = dir.clone();
let (granted, ungranted) = std::thread::spawn(move || {
let cav = Caveats {
exec: Scope::only(["cat".to_string()]),
..Caveats::top()
};
LandlockSandbox::new().apply(&cav).expect("apply landlock");
let granted = std::process::Command::new("cat")
.arg(dir_t.join("data.txt"))
.output();
let ungranted = std::process::Command::new("head")
.arg(dir_t.join("data.txt"))
.output();
(granted, ungranted)
})
.join()
.unwrap();
let granted = granted.expect("granted `cat` must still load and run");
assert!(
granted.status.success(),
"granted cat must succeed: {granted:?}"
);
assert_eq!(granted.stdout, b"payload\n");
let err = ungranted.expect_err("un-granted `head` must be exec-denied by Landlock");
assert_eq!(
err.kind(),
std::io::ErrorKind::PermissionDenied,
"the denial must come from the kernel (EACCES on execve)"
);
let _ = fs::remove_dir_all(&dir);
}
#[test]
fn exec_escape_attempts_are_all_denied() {
use std::os::unix::fs::{symlink, PermissionsExt};
if skip_proof_unless_landlock() {
return;
}
let scratch = unique_dir("exec-escape"); fs::write(scratch.join("data.txt"), b"ok\n").unwrap();
let payload = scratch.join("payload");
if let Ok(src) = std::fs::read("/bin/cat").or_else(|_| std::fs::read("/usr/bin/cat")) {
fs::write(&payload, src).unwrap();
fs::set_permissions(&payload, std::fs::Permissions::from_mode(0o755)).unwrap();
}
let script = scratch.join("script.sh");
fs::write(&script, b"#!/bin/sh\necho pwned\n").unwrap();
fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
let link = scratch.join("sh-link");
let _ = symlink("/bin/sh", &link);
let lib_execs: Vec<PathBuf> = [
"/usr/lib/klibc/bin/sh",
"/usr/lib/initramfs-tools/bin/busybox",
"/usr/lib/git-core/git",
]
.iter()
.map(PathBuf::from)
.filter(|p| p.exists())
.collect();
let scratch_t = scratch.clone();
let (attempts, control) = std::thread::spawn(move || {
let cav = Caveats {
exec: Scope::only(["cat".to_string()]),
fs_write: Scope::only([scratch_t.to_string_lossy().into_owned()]),
..Caveats::top()
};
LandlockSandbox::new().apply(&cav).expect("apply landlock");
let mut attempts = vec![
(
"ungranted-tool".to_string(),
std::process::Command::new("head")
.arg("/etc/hostname")
.output(),
),
(
"written-payload".to_string(),
std::process::Command::new(scratch_t.join("payload")).output(),
),
(
"shebang-script".to_string(),
std::process::Command::new(scratch_t.join("script.sh")).output(),
),
(
"symlink-to-sh".to_string(),
std::process::Command::new(scratch_t.join("sh-link"))
.arg("-c")
.arg("echo pwned")
.output(),
),
];
for p in &lib_execs {
attempts.push((
format!("under-usr-lib:{}", p.display()),
std::process::Command::new(p).arg("--version").output(),
));
}
let control = std::process::Command::new("cat")
.arg(scratch_t.join("data.txt"))
.output();
(attempts, control)
})
.join()
.unwrap();
for (label, res) in attempts {
match res {
Err(e) => assert_eq!(
e.kind(),
std::io::ErrorKind::PermissionDenied,
"escape `{label}` failed for the wrong reason: {e:?}"
),
Ok(out) => panic!(
"escape `{label}` was NOT denied — it ran (status {:?}, stdout {:?})",
out.status, out.stdout
),
}
}
let control = control.expect("granted `cat` must still run");
assert!(
control.status.success() && control.stdout == b"ok\n",
"control: {control:?}"
);
let _ = fs::remove_dir_all(&scratch);
}
#[test]
fn read_base_excludes_bin_dirs_when_exec_confined() {
if skip_proof_unless_landlock() {
return;
}
let dir = unique_dir("read-narrow");
fs::write(dir.join("data.txt"), b"payload\n").unwrap();
let dir_t = dir.clone();
let (granted, head_bytes) = std::thread::spawn(move || {
let cav = Caveats {
exec: Scope::only(["cat".to_string()]),
fs_read: Scope::only([dir_t.to_string_lossy().into_owned()]),
..Caveats::top()
};
LandlockSandbox::new().apply(&cav).expect("apply landlock");
let granted = std::process::Command::new("cat")
.arg(dir_t.join("data.txt"))
.output();
let head_bytes = std::fs::read("/usr/bin/head").or_else(|_| std::fs::read("/bin/head"));
(granted, head_bytes)
})
.join()
.unwrap();
let granted = granted.expect("granted `cat` must load + run under narrowed reads");
assert!(
granted.status.success() && granted.stdout == b"payload\n",
"granted cat under narrowed reads: {granted:?}"
);
assert!(
head_bytes.is_err(),
"an un-granted bin-dir binary must be unreadable (trampoline corpus shrunk): {head_bytes:?}"
);
let _ = fs::remove_dir_all(&dir);
}
}
#[cfg(all(target_os = "macos", feature = "macos-seatbelt", test))]
mod seatbelt_kernel_tests {
use super::*;
use crate::Scope;
use std::fs;
use std::path::PathBuf;
#[derive(Debug, PartialEq, Eq)]
enum ProofGate {
Run,
Skip,
Fail,
}
fn proof_gate(supported: bool, required: bool) -> ProofGate {
match (supported, required) {
(true, _) => ProofGate::Run,
(false, true) => ProofGate::Fail,
(false, false) => ProofGate::Skip,
}
}
fn skip_proof_unless_seatbelt() -> bool {
let required = std::env::var("BRIDLE_REQUIRE_SEATBELT")
.map(|v| !v.is_empty() && v != "0")
.unwrap_or(false);
match proof_gate(seatbelt_is_supported(), required) {
ProofGate::Run => false,
ProofGate::Skip => {
eprintln!(
"skipping Seatbelt proof: /usr/bin/sandbox-exec unavailable \
(set BRIDLE_REQUIRE_SEATBELT=1 to require it, as macOS CI does)"
);
true
}
ProofGate::Fail => panic!(
"BRIDLE_REQUIRE_SEATBELT is set but /usr/bin/sandbox-exec is unavailable — \
the fs_write/fs_read kernel-enforcement proofs cannot be verified"
),
}
}
fn unique_dir(tag: &str) -> PathBuf {
use std::sync::atomic::{AtomicU64, Ordering};
static N: AtomicU64 = AtomicU64::new(0);
let mut d = std::env::temp_dir();
d.push(format!(
"agent-bridle-sb-{}-{}-{}",
tag,
std::process::id(),
N.fetch_add(1, Ordering::Relaxed)
));
fs::create_dir_all(&d).unwrap();
d
}
fn run_wrapped(cav: &Caveats, program: &str, args: &[&str]) -> std::process::ExitStatus {
let prefix = SeatbeltSandbox::new()
.command_prefix(cav)
.expect("a restricted axis must yield a wrapper prefix");
assert!(!prefix.is_empty(), "expected a sandbox-exec wrapper");
std::process::Command::new(&prefix[0])
.args(&prefix[1..])
.arg(program)
.args(args)
.status()
.expect("spawn sandbox-exec")
}
#[test]
fn proof_gate_required_but_unsupported_is_a_failure() {
assert_eq!(proof_gate(true, false), ProofGate::Run);
assert_eq!(proof_gate(true, true), ProofGate::Run);
assert_eq!(proof_gate(false, false), ProofGate::Skip);
assert_eq!(proof_gate(false, true), ProofGate::Fail);
}
#[test]
fn fs_write_is_kernel_enforced_outside_scope_denied_inside_allowed() {
if skip_proof_unless_seatbelt() {
return;
}
let allowed = unique_dir("w-allowed");
let forbidden = unique_dir("w-forbidden");
let cav = Caveats {
fs_write: Scope::only([allowed.to_string_lossy().into_owned()]),
..Caveats::top()
};
let inside = run_wrapped(
&cav,
"/usr/bin/touch",
&[allowed.join("ok.txt").to_str().unwrap()],
);
assert!(
inside.success(),
"writing within fs_write scope must succeed"
);
assert!(
allowed.join("ok.txt").exists(),
"the in-scope file must exist"
);
let outside = run_wrapped(
&cav,
"/usr/bin/touch",
&[forbidden.join("escape.txt").to_str().unwrap()],
);
assert!(
!outside.success(),
"the kernel must deny a write outside fs_write scope"
);
assert!(
!forbidden.join("escape.txt").exists(),
"the out-of-scope file must NOT have been created"
);
let _ = fs::remove_dir_all(&allowed);
let _ = fs::remove_dir_all(&forbidden);
}
#[test]
fn empty_fs_write_scope_denies_all_writes() {
if skip_proof_unless_seatbelt() {
return;
}
let dir = unique_dir("w-none");
let cav = Caveats {
fs_write: Scope::none(),
..Caveats::top()
};
let target = dir.join("x.txt");
let prefix = SeatbeltSandbox::new().command_prefix(&cav).expect("prefix");
let out = std::process::Command::new(&prefix[0])
.args(&prefix[1..])
.arg("/usr/bin/touch")
.arg(&target)
.output()
.expect("spawn sandbox-exec");
assert!(!out.status.success(), "empty fs_write must deny all writes");
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
stderr.contains("Operation not permitted"),
"denial must be a sandbox EPERM, got: {stderr:?}"
);
assert!(!target.exists());
let _ = fs::remove_dir_all(&dir);
}
#[test]
fn fs_read_is_kernel_enforced_outside_scope_denied_inside_allowed() {
if skip_proof_unless_seatbelt() {
return;
}
let allowed = unique_dir("r-allowed");
let forbidden = unique_dir("r-forbidden");
fs::write(allowed.join("ok.txt"), b"in-scope").unwrap();
fs::write(forbidden.join("secret.txt"), b"out-of-scope").unwrap();
let cav = Caveats {
fs_read: Scope::only([allowed.to_string_lossy().into_owned()]),
..Caveats::top()
};
let inside = run_wrapped(
&cav,
"/bin/cat",
&[allowed.join("ok.txt").to_str().unwrap()],
);
assert!(
inside.success(),
"in-scope cat must load and read under read-confinement"
);
let outside = run_wrapped(
&cav,
"/bin/cat",
&[forbidden.join("secret.txt").to_str().unwrap()],
);
assert!(
!outside.success(),
"reading outside fs_read scope must be kernel-denied"
);
let _ = fs::remove_dir_all(&allowed);
let _ = fs::remove_dir_all(&forbidden);
}
#[test]
fn net_fully_denied_kernel_blocks_egress() {
if skip_proof_unless_seatbelt() {
return;
}
let curl = "/usr/bin/curl";
if !std::path::Path::new(curl).exists() {
eprintln!("skipping: no curl(1) on this host");
return;
}
let cav = Caveats {
net: Scope::none(),
..Caveats::top()
};
let benign = run_wrapped(&cav, "/bin/echo", &["ok"]);
assert!(
benign.success(),
"net:none must still allow non-network commands (profile must parse)"
);
let confined = run_wrapped(&cav, curl, &["-sS", "--max-time", "5", "http://1.1.1.1/"]);
assert_eq!(
confined.code(),
Some(7),
"egress under net:none must be kernel-denied at the socket (curl exit 7)"
);
}
fn spawn_loopback_http(bind: &str) -> Option<std::net::SocketAddr> {
let listener = std::net::TcpListener::bind(bind).ok()?;
let addr = listener.local_addr().ok()?;
std::thread::spawn(move || {
if let Ok((mut sock, _)) = listener.accept() {
use std::io::{Read, Write};
let mut buf = [0u8; 1024];
let _ = sock.read(&mut buf);
let _ = sock.write_all(b"HTTP/1.0 200 OK\r\nContent-Length: 2\r\n\r\nok");
}
});
Some(addr)
}
#[test]
fn net_loopback_only_permits_loopback_interface_denies_offbox() {
if skip_proof_unless_seatbelt() {
return;
}
let curl = "/usr/bin/curl";
if !std::path::Path::new(curl).exists() {
eprintln!("skipping: no curl(1) on this host");
return;
}
let v4 = spawn_loopback_http("127.0.0.1:0").expect("bind v4 loopback");
let cav = Caveats {
net: Scope::only(["127.0.0.1".to_string()]),
..Caveats::top()
};
assert!(
run_wrapped(&cav, "/bin/echo", &["ok"]).success(),
"loopback-only profile must still run non-network commands (must parse)"
);
let v4_url = format!("http://127.0.0.1:{}/", v4.port());
assert!(
run_wrapped(&cav, curl, &["-sS", "--max-time", "5", &v4_url]).success(),
"net:Only([127.0.0.1]) must kernel-PERMIT v4 loopback egress"
);
if let Some(v6) = spawn_loopback_http("[::1]:0") {
let v6_url = format!("http://[::1]:{}/", v6.port());
assert!(
run_wrapped(&cav, curl, &["-sS", "--max-time", "5", &v6_url]).success(),
"net:Only([127.0.0.1]) kernel-permits the whole loopback interface, incl. ::1 (ADR 0015 D2)"
);
}
let offbox = run_wrapped_output(
&cav,
curl,
&["-sS", "-v", "--max-time", "5", "http://1.1.1.1/"],
);
assert_eq!(
offbox.status.code(),
Some(7),
"net:Only([127.0.0.1]) must kernel-DENY off-box egress (curl exit 7)"
);
let stderr = String::from_utf8_lossy(&offbox.stderr);
assert!(
stderr.contains("Operation not permitted"),
"off-box denial must be a kernel EPERM, not a routing failure: {stderr}"
);
}
fn run_wrapped_output(cav: &Caveats, program: &str, args: &[&str]) -> std::process::Output {
let prefix = SeatbeltSandbox::new()
.command_prefix(cav)
.expect("a restricted axis must yield a wrapper prefix");
assert!(!prefix.is_empty(), "expected a sandbox-exec wrapper");
std::process::Command::new(&prefix[0])
.args(&prefix[1..])
.arg(program)
.args(args)
.output()
.expect("spawn sandbox-exec")
}
#[test]
fn exec_allowlist_permits_listed_denies_unlisted_child() {
if skip_proof_unless_seatbelt() {
return;
}
let cav = Caveats {
exec: Scope::only(["/bin/zsh".to_string(), "/usr/bin/true".to_string()]),
..Caveats::top()
};
let out = run_wrapped_output(
&cav,
"/bin/zsh",
&["-c", "/usr/bin/true; echo T=$?; /usr/bin/false; echo F=$?"],
);
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
stdout.contains("T=0"),
"a listed binary must exec and run (T=0): {stdout:?}"
);
assert!(
stdout.contains("F=127"),
"an unlisted binary must be kernel-denied at EXEC (status 127), not run: {stdout:?}"
);
}
#[test]
fn granted_shell_cannot_exec_any_unlisted_child() {
if skip_proof_unless_seatbelt() {
return;
}
let cav = Caveats {
exec: Scope::only(["/bin/zsh".to_string()]),
..Caveats::top()
};
let out = run_wrapped_output(&cav, "/bin/zsh", &["-c", "/usr/bin/true; echo S=$?"]);
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
stdout.contains("S=127"),
"a shell granted only itself must be denied every child exec (S=127): {stdout:?}"
);
}
#[test]
fn granted_interpreter_cannot_trampoline_to_unlisted_binary() {
if skip_proof_unless_seatbelt() {
return;
}
let cav = Caveats {
exec: Scope::only(["/usr/bin/perl".to_string()]),
..Caveats::top()
};
let script = "print \"PERL-RAN\\n\"; \
exec(\"/usr/bin/true\"); print \"DIRECT-DENIED\\n\"; \
exec(\"/usr/lib/dyld\", \"/usr/bin/true\"); print \"TRAMPOLINE-DENIED\\n\";";
let out = run_wrapped_output(&cav, "/usr/bin/perl", &["-e", script]);
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
stdout.contains("PERL-RAN"),
"the granted interpreter must run: {stdout:?}"
);
assert!(
stdout.contains("DIRECT-DENIED"),
"direct exec of an unlisted binary must be denied: {stdout:?}"
);
assert!(
stdout.contains("TRAMPOLINE-DENIED"),
"the dyld loader trampoline must be denied (no standing loader entry): {stdout:?}"
);
}
#[test]
fn exec_confinement_does_not_break_dynamic_linking() {
if skip_proof_unless_seatbelt() {
return;
}
let curl = "/usr/bin/curl";
if !std::path::Path::new(curl).exists() {
eprintln!("skipping: no curl(1) on this host");
return;
}
let cav = Caveats {
exec: Scope::only([curl.to_string()]),
..Caveats::top()
};
let status = run_wrapped(&cav, curl, &["--version"]);
assert!(
status.success(),
"an allow-listed dynamic binary must load + run under exec confinement"
);
}
}