use std::net::IpAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use murmur_artifact::security_warnings::{
security_warning_link, W_SEC_001, W_SEC_002, W_SEC_005, W_SEC_010,
};
use murmur_artifact::InterpreterRuntimeGrant;
use crate::types::CapabilityPolicy;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum EnforcementTier {
KernelSealed,
KernelFull,
KernelSeccompOnly,
EnvironmentOnly,
}
pub(crate) fn tier_from_probe(
is_linux: bool,
landlock_fully_enforced: Option<bool>,
sealed: crate::sealed::SealedProbe,
) -> EnforcementTier {
if !is_linux {
return EnforcementTier::EnvironmentOnly;
}
match landlock_fully_enforced {
Some(true)
if sealed.userns_grant.permits_userns()
&& sealed.namespace == crate::sealed::NamespaceProbe::Ok =>
{
EnforcementTier::KernelSealed
}
Some(true) => EnforcementTier::KernelFull,
Some(false) | None => EnforcementTier::KernelSeccompOnly,
}
}
pub(crate) fn applied_tier(
host_tier: EnforcementTier,
declared: murmur_artifact::ContainmentClass,
) -> EnforcementTier {
match host_tier {
EnforcementTier::KernelSealed if declared < murmur_artifact::ContainmentClass::Sealed => {
EnforcementTier::KernelFull
}
other => other,
}
}
#[derive(Debug, Clone, Copy)]
pub struct HostProbe {
landlock_fully_enforced: Option<bool>,
sealed: crate::sealed::SealedProbe,
}
static PROBES_TAKEN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
impl HostProbe {
pub fn probe() -> HostProbe {
PROBES_TAKEN.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
#[cfg(target_os = "linux")]
let probe = HostProbe {
landlock_fully_enforced: linux_enforce::probe_landlock_full_access(),
sealed: crate::sealed::probe_sealed_support(),
};
#[cfg(not(target_os = "linux"))]
let probe = HostProbe {
landlock_fully_enforced: None,
sealed: crate::sealed::SealedProbe::default(),
};
probe
}
#[cfg(test)]
pub(crate) fn from_parts(
landlock_fully_enforced: Option<bool>,
sealed: crate::sealed::SealedProbe,
) -> HostProbe {
HostProbe {
landlock_fully_enforced,
sealed,
}
}
pub(crate) fn tier(&self) -> EnforcementTier {
tier_from_probe(
cfg!(target_os = "linux"),
self.landlock_fully_enforced,
self.sealed,
)
}
pub fn sealed_blocker(&self) -> Option<crate::sealed::SealedBlocker> {
crate::sealed::sealed_blocker(
cfg!(target_os = "linux"),
self.landlock_fully_enforced == Some(true),
self.sealed,
)
}
pub fn userns_grant(&self) -> Option<crate::sealed::UsernsGrant> {
if cfg!(target_os = "linux") {
Some(self.sealed.userns_grant)
} else {
None
}
}
pub fn probes_taken() -> u64 {
PROBES_TAKEN.load(std::sync::atomic::Ordering::Relaxed)
}
}
pub(crate) fn detect_enforcement_tier() -> EnforcementTier {
HostProbe::probe().tier()
}
pub fn detect_sealed_blocker() -> Option<crate::sealed::SealedBlocker> {
HostProbe::probe().sealed_blocker()
}
pub fn detect_userns_grant() -> Option<crate::sealed::UsernsGrant> {
HostProbe::probe().userns_grant()
}
pub(crate) fn resolve_network_allowlist_ips(
network_allow: &[String],
) -> Result<Vec<IpAddr>, String> {
let rules = crate::network_policy::parse_network_allow_rules(network_allow)
.map_err(|error| error.to_string())?;
let mut ips = std::collections::BTreeSet::new();
for rule in &rules {
let host = rule.host.as_str();
match crate::dns_resolver::resolve(host) {
crate::dns_resolver::Resolution::Resolved(addresses) => ips.extend(addresses),
crate::dns_resolver::Resolution::DoesNotExist => {}
crate::dns_resolver::Resolution::DidNotAnswer(reason) => crate::runtime_err!(
"[capsule-runtime] warning: the network allowlist host '{host}' could not be \
resolved at launch: {reason} (a subprocess reaching it by literal address is \
denied for this run)"
),
}
}
Ok(ips.into_iter().collect())
}
pub(crate) fn resolve_exec_allowlist(shell_allow: &[String]) -> Vec<PathBuf> {
let path_dirs: Vec<PathBuf> = std::env::var_os("PATH")
.map(|path| std::env::split_paths(&path).collect())
.unwrap_or_default();
resolve_exec_allowlist_in(shell_allow, &path_dirs)
}
fn resolve_exec_allowlist_in(shell_allow: &[String], path_dirs: &[PathBuf]) -> Vec<PathBuf> {
let mut resolved = std::collections::BTreeSet::new();
for entry in shell_allow {
if entry.is_empty() {
continue;
}
if entry.contains('/') {
if let Ok(canonical) = std::fs::canonicalize(entry) {
resolved.insert(canonical);
}
continue;
}
for dir in path_dirs {
let candidate = dir.join(entry);
if is_executable_file(&candidate) {
if let Ok(canonical) = std::fs::canonicalize(&candidate) {
resolved.insert(canonical);
}
}
}
}
resolved.into_iter().collect()
}
pub(crate) fn resolve_invoked_binary_path(binary: &str) -> String {
let path_dirs: Vec<PathBuf> = std::env::var_os("PATH")
.map(|path| std::env::split_paths(&path).collect())
.unwrap_or_default();
resolve_invoked_binary_path_in(binary, &path_dirs)
}
pub(crate) fn resolve_invoked_binary_path_in(binary: &str, path_dirs: &[PathBuf]) -> String {
let fallback = || binary.to_string();
if binary.is_empty() {
return fallback();
}
if binary.contains('/') {
return match std::fs::canonicalize(binary) {
Ok(canonical) => canonical.to_string_lossy().into_owned(),
Err(_) => fallback(),
};
}
for dir in path_dirs {
let candidate = dir.join(binary);
if is_executable_file(&candidate) {
if let Ok(canonical) = std::fs::canonicalize(&candidate) {
return canonical.to_string_lossy().into_owned();
}
}
}
fallback()
}
pub fn find_on_path(binary: &str) -> Option<PathBuf> {
std::env::split_paths(&std::env::var_os("PATH")?)
.map(|dir| dir.join(binary))
.find(|candidate| is_executable_file(candidate))
}
pub(crate) fn is_executable_file(path: &Path) -> bool {
use std::os::unix::fs::PermissionsExt;
std::fs::metadata(path)
.map(|meta| meta.is_file() && meta.permissions().mode() & 0o111 != 0)
.unwrap_or(false)
}
const DEFAULT_LIBRARY_SEARCH_DIRS: &[&str] = &[
"/lib",
"/lib64",
"/usr/lib",
"/usr/lib64",
"/lib/x86_64-linux-gnu",
"/usr/lib/x86_64-linux-gnu",
"/lib/aarch64-linux-gnu",
"/usr/lib/aarch64-linux-gnu",
];
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct ElfDependencies {
pub(crate) interp: Option<String>,
pub(crate) needed: Vec<String>,
pub(crate) runpaths: Vec<String>,
}
fn read_u16(bytes: &[u8], off: usize, le: bool) -> Option<u16> {
let raw: [u8; 2] = bytes.get(off..off + 2)?.try_into().ok()?;
Some(if le {
u16::from_le_bytes(raw)
} else {
u16::from_be_bytes(raw)
})
}
fn read_u32(bytes: &[u8], off: usize, le: bool) -> Option<u32> {
let raw: [u8; 4] = bytes.get(off..off + 4)?.try_into().ok()?;
Some(if le {
u32::from_le_bytes(raw)
} else {
u32::from_be_bytes(raw)
})
}
fn read_u64(bytes: &[u8], off: usize, le: bool) -> Option<u64> {
let raw: [u8; 8] = bytes.get(off..off + 8)?.try_into().ok()?;
Some(if le {
u64::from_le_bytes(raw)
} else {
u64::from_be_bytes(raw)
})
}
fn cstr_at_start(region: &[u8]) -> String {
let nul = region.iter().position(|&b| b == 0).unwrap_or(region.len());
String::from_utf8_lossy(®ion[..nul]).into_owned()
}
fn vaddr_to_file_offset(vaddr: u64, loads: &[(u64, u64, u64)]) -> Option<u64> {
for &(seg_vaddr, seg_off, seg_filesz) in loads {
if vaddr >= seg_vaddr && vaddr < seg_vaddr.checked_add(seg_filesz)? {
return Some(seg_off + (vaddr - seg_vaddr));
}
}
None
}
pub(crate) fn parse_elf_dependencies(bytes: &[u8]) -> Option<ElfDependencies> {
if bytes.len() < 64 || &bytes[0..4] != b"\x7fELF" {
return None;
}
if bytes[4] != 2 {
return None;
}
let le = match bytes[5] {
1 => true,
2 => false,
_ => return None,
};
let e_phoff = read_u64(bytes, 32, le)? as usize;
let e_phentsize = read_u16(bytes, 54, le)? as usize;
let e_phnum = read_u16(bytes, 56, le)? as usize;
if e_phentsize < 56 {
return None;
}
let mut interp: Option<String> = None;
let mut dynamic_region: Option<(u64, u64)> = None; let mut loads: Vec<(u64, u64, u64)> = Vec::new();
for i in 0..e_phnum {
let base = e_phoff.checked_add(i.checked_mul(e_phentsize)?)?;
let p_type = read_u32(bytes, base, le)?;
let p_offset = read_u64(bytes, base + 8, le)?;
let p_vaddr = read_u64(bytes, base + 16, le)?;
let p_filesz = read_u64(bytes, base + 32, le)?;
match p_type {
1 => loads.push((p_vaddr, p_offset, p_filesz)), 2 => dynamic_region = Some((p_offset, p_filesz)), 3 => {
let start = p_offset as usize;
let end = start.checked_add(p_filesz as usize)?;
interp = Some(cstr_at_start(bytes.get(start..end)?));
}
_ => {}
}
}
let Some((dyn_off, dyn_size)) = dynamic_region else {
return Some(ElfDependencies {
interp,
needed: Vec::new(),
runpaths: Vec::new(),
});
};
let mut needed_offsets: Vec<u64> = Vec::new();
let mut runpath_offsets: Vec<u64> = Vec::new();
let mut strtab_vaddr: Option<u64> = None;
let mut strsz: Option<u64> = None;
let mut off = dyn_off as usize;
let dyn_end = off.checked_add(dyn_size as usize)?;
while off + 16 <= dyn_end {
let d_tag = read_u64(bytes, off, le)?;
let d_val = read_u64(bytes, off + 8, le)?;
match d_tag {
0 => break, 1 => needed_offsets.push(d_val), 5 => strtab_vaddr = Some(d_val), 10 => strsz = Some(d_val), 15 | 29 => runpath_offsets.push(d_val), _ => {}
}
off += 16;
}
let mut needed = Vec::new();
let mut runpaths = Vec::new();
if let (Some(strtab_vaddr), Some(strsz)) = (strtab_vaddr, strsz) {
if let Some(strtab_off) = vaddr_to_file_offset(strtab_vaddr, &loads) {
let start = strtab_off as usize;
let strtab = start
.checked_add(strsz as usize)
.and_then(|end| bytes.get(start..end))
.or_else(|| bytes.get(start..));
if let Some(strtab) = strtab {
for offset in needed_offsets {
if let Some(region) = strtab.get(offset as usize..) {
needed.push(cstr_at_start(region));
}
}
for offset in runpath_offsets {
if let Some(region) = strtab.get(offset as usize..) {
for part in cstr_at_start(region).split(':') {
if !part.is_empty() {
runpaths.push(part.to_string());
}
}
}
}
}
}
}
Some(ElfDependencies {
interp,
needed,
runpaths,
})
}
fn expand_origin(rpath: &str, origin_dir: &Path) -> PathBuf {
if rpath.contains("$ORIGIN") {
let origin = origin_dir.to_string_lossy();
PathBuf::from(
rpath
.replace("${ORIGIN}", &origin)
.replace("$ORIGIN", &origin),
)
} else {
PathBuf::from(rpath)
}
}
pub(crate) fn resolve_soname(
soname: &str,
origin_dir: &Path,
runpaths: &[String],
search_dirs: &[PathBuf],
) -> Option<PathBuf> {
if soname.is_empty() {
return None;
}
if soname.contains('/') {
let candidate = if Path::new(soname).is_absolute() {
PathBuf::from(soname)
} else {
origin_dir.join(soname)
};
return std::fs::canonicalize(&candidate).ok();
}
for rpath in runpaths {
let candidate = expand_origin(rpath, origin_dir).join(soname);
if candidate.is_file() {
if let Ok(canonical) = std::fs::canonicalize(&candidate) {
return Some(canonical);
}
}
}
for dir in search_dirs {
let candidate = dir.join(soname);
if candidate.is_file() {
if let Ok(canonical) = std::fs::canonicalize(&candidate) {
return Some(canonical);
}
}
}
None
}
pub(crate) fn resolve_landlock_grants(exec_allow_paths: &[PathBuf]) -> Vec<PathBuf> {
let search_dirs: Vec<PathBuf> = DEFAULT_LIBRARY_SEARCH_DIRS
.iter()
.map(PathBuf::from)
.collect();
resolve_landlock_grants_in(exec_allow_paths, &search_dirs)
}
fn resolve_landlock_grants_in(
exec_allow_paths: &[PathBuf],
search_dirs: &[PathBuf],
) -> Vec<PathBuf> {
let mut grants: std::collections::BTreeSet<PathBuf> = std::collections::BTreeSet::new();
let mut visited: std::collections::BTreeSet<PathBuf> = std::collections::BTreeSet::new();
let mut queue: std::collections::VecDeque<PathBuf> = std::collections::VecDeque::new();
for path in exec_allow_paths {
if let Ok(canonical) = std::fs::canonicalize(path) {
queue.push_back(canonical);
}
}
while let Some(path) = queue.pop_front() {
if !visited.insert(path.clone()) {
continue;
}
grants.insert(path.clone());
let Ok(bytes) = std::fs::read(&path) else {
continue;
};
let Some(deps) = parse_elf_dependencies(&bytes) else {
continue;
};
let origin_dir = path.parent().unwrap_or_else(|| Path::new("/"));
if let Some(interp) = &deps.interp {
if let Ok(canonical) = std::fs::canonicalize(interp) {
queue.push_back(canonical);
}
}
for soname in &deps.needed {
if let Some(resolved) = resolve_soname(soname, origin_dir, &deps.runpaths, search_dirs)
{
queue.push_back(resolved);
}
}
}
grants.into_iter().collect()
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct LandlockGrant {
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) path: PathBuf,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) list_dir: bool,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) executable: bool,
}
impl LandlockGrant {
fn non_listable_files(paths: Vec<PathBuf>) -> Vec<LandlockGrant> {
paths
.into_iter()
.map(|path| LandlockGrant {
path,
list_dir: false,
executable: true,
})
.collect()
}
}
pub(crate) fn resolve_interpreter_runtime_grants(
grants: &[InterpreterRuntimeGrant],
) -> Vec<LandlockGrant> {
grants
.iter()
.flat_map(|grant| {
grant.dirs.iter().map(|dir| LandlockGrant {
path: PathBuf::from(&dir.path),
list_dir: dir.list_dir,
executable: true,
})
})
.collect()
}
pub(crate) fn resolve_sealed_bind_dirs(
exec_allow_paths: &[PathBuf],
policy: &CapabilityPolicy,
) -> Vec<PathBuf> {
let fixed: Vec<&Path> = crate::sealed::SEALED_RUNTIME_PATHS
.iter()
.map(Path::new)
.collect();
let mut dirs: Vec<PathBuf> = Vec::new();
let candidates = exec_allow_paths
.iter()
.filter_map(|binary| binary.parent().map(Path::to_path_buf))
.chain(
policy
.shell_interpreter_runtime
.iter()
.flat_map(|grant| grant.dirs.iter().map(|dir| PathBuf::from(&dir.path))),
);
for candidate in candidates {
if !candidate.is_absolute() {
continue;
}
if fixed.iter().any(|root| candidate.starts_with(root)) {
continue;
}
if dirs.contains(&candidate) {
continue;
}
dirs.push(candidate);
}
dirs
}
pub(crate) fn resolve_staged_runtime_dirs(policy: &CapabilityPolicy) -> Vec<PathBuf> {
let mut dirs: Vec<PathBuf> = Vec::new();
for grant in &policy.shell_staged_runtime {
let path = PathBuf::from(&grant.source_path);
if !path.is_absolute() || dirs.contains(&path) {
continue;
}
dirs.push(path);
}
dirs
}
pub(crate) fn resolve_staged_runtime_landlock_grants(
policy: &CapabilityPolicy,
) -> Vec<LandlockGrant> {
resolve_staged_runtime_dirs(policy)
.into_iter()
.map(|path| LandlockGrant {
path,
list_dir: true,
executable: true,
})
.collect()
}
pub(crate) fn resolve_sealed_runtime_landlock_grants(tier: EnforcementTier) -> Vec<LandlockGrant> {
if tier != EnforcementTier::KernelSealed {
return Vec::new();
}
crate::sealed::SEALED_RUNTIME_PATHS
.iter()
.map(|path| LandlockGrant {
path: PathBuf::from(path),
list_dir: true,
executable: false,
})
.collect()
}
pub(crate) fn resolve_sealed_etc_landlock_grants(tier: EnforcementTier) -> Vec<LandlockGrant> {
if tier != EnforcementTier::KernelSealed {
return Vec::new();
}
crate::sealed::SEALED_ETC_PATHS
.iter()
.map(|entry| LandlockGrant {
path: PathBuf::from(entry.path),
list_dir: entry.directory,
executable: false,
})
.collect()
}
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct CapsuleDeviceGrant {
pub(crate) path: &'static str,
pub(crate) writable: bool,
}
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) const CAPSULE_DEVICE_GRANTS: &[CapsuleDeviceGrant] = &[
CapsuleDeviceGrant {
path: "/dev/null",
writable: true,
},
CapsuleDeviceGrant {
path: "/dev/zero",
writable: false,
},
CapsuleDeviceGrant {
path: "/dev/urandom",
writable: false,
},
];
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) fn network_ip_allowed(ip: IpAddr, network_allow_ips: &[IpAddr]) -> bool {
network_allow_ips.contains(&ip)
}
const LINUX_AF_UNIX: i32 = 1;
const LINUX_AF_INET: i32 = 2;
const LINUX_AF_INET6: i32 = 10;
const LINUX_AF_NETLINK: i32 = 16;
const LINUX_AF_PACKET: i32 = 17;
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) fn denied_socket_domains(unix_sockets_allowed: bool) -> Vec<i32> {
let mut denied = vec![LINUX_AF_NETLINK, LINUX_AF_PACKET];
if !unix_sockets_allowed {
denied.push(LINUX_AF_UNIX);
}
denied
}
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) fn allowed_socket_domains(unix_sockets_allowed: bool) -> Vec<i32> {
let mut allowed = vec![LINUX_AF_INET_DOMAIN, LINUX_AF_INET6_DOMAIN];
if unix_sockets_allowed {
allowed.push(LINUX_AF_UNIX);
}
allowed
}
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const LINUX_AF_INET_DOMAIN: i32 = LINUX_AF_INET;
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const LINUX_AF_INET6_DOMAIN: i32 = LINUX_AF_INET6;
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) const SECCOMP_SYSCALL_ALLOWLIST: &[&str] = &[
"execve",
"execveat",
"clone",
"clone3",
"fork",
"vfork",
"exit",
"exit_group",
"wait4",
"waitid",
"kill",
"tkill",
"tgkill",
"rt_sigaction",
"rt_sigprocmask",
"rt_sigreturn",
"rt_sigsuspend",
"rt_sigpending",
"rt_sigtimedwait",
"rt_sigqueueinfo",
"rt_tgsigqueueinfo",
"sigaltstack",
"restart_syscall",
"pause",
"set_tid_address",
"set_robust_list",
"get_robust_list",
"rseq",
"prctl",
"arch_prctl",
"getpid",
"getppid",
"gettid",
"getpgrp",
"getpgid",
"setpgid",
"getsid",
"setsid",
"getpriority",
"setpriority",
"ioprio_get",
"ioprio_set",
"sched_yield",
"sched_getaffinity",
"sched_setaffinity",
"sched_getparam",
"sched_setparam",
"sched_getscheduler",
"sched_setscheduler",
"sched_getattr",
"sched_setattr",
"sched_get_priority_max",
"sched_get_priority_min",
"sched_rr_get_interval",
"getcpu",
"capget",
"capset",
"prlimit64",
"getrlimit",
"setrlimit",
"getrusage",
"getitimer",
"setitimer",
"times",
"uname",
"sysinfo",
"pidfd_open",
"pidfd_send_signal",
"mmap",
"munmap",
"mprotect",
"mremap",
"madvise",
"brk",
"mlock",
"mlock2",
"munlock",
"mlockall",
"munlockall",
"membarrier",
"msync",
"mincore",
"memfd_create",
"pkey_alloc",
"pkey_free",
"pkey_mprotect",
"shmget",
"shmat",
"shmdt",
"shmctl",
"open",
"openat",
"openat2",
"creat",
"close",
"close_range",
"read",
"write",
"pread64",
"pwrite64",
"readv",
"writev",
"preadv",
"pwritev",
"preadv2",
"pwritev2",
"lseek",
"dup",
"dup2",
"dup3",
"fcntl",
"flock",
"fsync",
"fdatasync",
"sync",
"syncfs",
"sync_file_range",
"fadvise64",
"readahead",
"truncate",
"ftruncate",
"fallocate",
"stat",
"fstat",
"lstat",
"newfstatat",
"statx",
"access",
"faccessat",
"faccessat2",
"getdents",
"getdents64",
"getcwd",
"chdir",
"fchdir",
"mkdir",
"mkdirat",
"rmdir",
"unlink",
"unlinkat",
"rename",
"renameat",
"renameat2",
"link",
"linkat",
"symlink",
"symlinkat",
"readlink",
"readlinkat",
"chmod",
"fchmod",
"fchmodat",
"fchmodat2",
"chown",
"fchown",
"fchownat",
"lchown",
"umask",
"utime",
"utimes",
"utimensat",
"futimesat",
"mknod",
"mknodat",
"sendfile",
"copy_file_range",
"splice",
"tee",
"vmsplice",
"ioctl",
"poll",
"ppoll",
"select",
"pselect6",
"epoll_create",
"epoll_create1",
"epoll_ctl",
"epoll_wait",
"epoll_pwait",
"epoll_pwait2",
"eventfd",
"eventfd2",
"pipe",
"pipe2",
"inotify_init",
"inotify_init1",
"inotify_add_watch",
"inotify_rm_watch",
"signalfd",
"signalfd4",
"timerfd_create",
"timerfd_settime",
"timerfd_gettime",
"getxattr",
"lgetxattr",
"fgetxattr",
"setxattr",
"lsetxattr",
"fsetxattr",
"listxattr",
"llistxattr",
"flistxattr",
"removexattr",
"lremovexattr",
"fremovexattr",
"statfs",
"fstatfs",
"name_to_handle_at",
"io_setup",
"io_destroy",
"io_submit",
"io_cancel",
"io_getevents",
"connect",
"sendto",
"socketpair",
"bind",
"listen",
"accept",
"accept4",
"getsockname",
"getpeername",
"getsockopt",
"setsockopt",
"shutdown",
"sendmsg",
"sendmmsg",
"recvfrom",
"recvmsg",
"recvmmsg",
"clock_gettime",
"clock_getres",
"clock_nanosleep",
"nanosleep",
"gettimeofday",
"time",
"timer_create",
"timer_settime",
"timer_gettime",
"timer_getoverrun",
"timer_delete",
"alarm",
"msgget",
"msgsnd",
"msgrcv",
"msgctl",
"semget",
"semop",
"semctl",
"semtimedop",
"mq_open",
"mq_unlink",
"mq_timedsend",
"mq_timedreceive",
"mq_notify",
"mq_getsetattr",
"getrandom",
"futex",
"futex_waitv",
"getuid",
"geteuid",
"getgid",
"getegid",
"setuid",
"setgid",
"setreuid",
"setregid",
"setresuid",
"setresgid",
"getresuid",
"getresgid",
"getgroups",
"setgroups",
"setfsuid",
"setfsgid",
];
#[allow(dead_code)]
pub(crate) const SECCOMP_MUST_STAY_DENIED: &[&str] = &[
"io_uring_setup",
"io_uring_enter",
"io_uring_register",
"bpf",
"userfaultfd",
"perf_event_open",
"process_vm_readv",
"process_vm_writev",
"open_by_handle_at",
"keyctl",
"add_key",
"request_key",
"ptrace",
"mount",
"umount2",
"pivot_root",
"chroot",
"unshare",
"setns",
"init_module",
"finit_module",
"delete_module",
"kexec_load",
"kexec_file_load",
"reboot",
"swapon",
"swapoff",
"acct",
"quotactl",
"nfsservctl",
"iopl",
"ioperm",
"vm86",
"vm86old",
"_sysctl",
"sysfs",
"uselib",
"create_module",
"get_kernel_syms",
"query_module",
"lookup_dcookie",
"clock_adjtime",
"clock_settime",
"settimeofday",
"stime",
"adjtimex",
"set_mempolicy",
"get_mempolicy",
"mbind",
"migrate_pages",
"move_pages",
"fanotify_init",
"fanotify_mark",
"seccomp",
"personality",
"kcmp",
];
#[allow(dead_code)]
pub(crate) const SECCOMP_FILTER_INSTRUCTION_BUDGET: usize = 512;
#[allow(dead_code)]
pub(crate) const BPF_MAXINSNS_CEILING: usize = 4096;
#[allow(dead_code)]
pub(crate) const SECCOMP_MAX_INSNS_PER_PATH: usize = 32768;
#[derive(Debug, Clone)]
pub(crate) struct ShellEnforcement {
pub(crate) tier: EnforcementTier,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) network_allow_ips: Vec<IpAddr>,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) network_allow_rules: Vec<crate::network_policy::NetworkAllowRule>,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) egress_tcp_ports: Vec<u16>,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) unix_sockets_allowed: bool,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) workdir_exec: bool,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) landlock_grants: Vec<LandlockGrant>,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) sealed_bind_dirs: Vec<PathBuf>,
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) staged_runtime_dirs: Vec<PathBuf>,
pub(crate) resource_limits: crate::resources::HostResourceLimits,
pub(crate) cgroup_scope: Option<Arc<crate::cgroup::CgroupScope>>,
pub(crate) workdir_guard: Option<Arc<crate::resources::WorkdirGuard>>,
}
impl ShellEnforcement {
pub(crate) fn resolve(
policy: &CapabilityPolicy,
declared: murmur_artifact::ContainmentClass,
probe: HostProbe,
) -> Result<Self, String> {
let tier = applied_tier(probe.tier(), declared);
let network_allow_ips = resolve_network_allowlist_ips(&policy.network_allow)?;
let network_allow_rules =
crate::network_policy::parse_network_allow_rules(&policy.network_allow)
.map_err(|error| error.to_string())?;
let egress_tcp_ports = crate::egress_proxy::egress_listen_ports(&network_allow_rules);
let exec_allow_paths = resolve_exec_allowlist(&policy.shell_allow);
let mut landlock_grants =
LandlockGrant::non_listable_files(resolve_landlock_grants(&exec_allow_paths));
landlock_grants.extend(resolve_interpreter_runtime_grants(
&policy.shell_interpreter_runtime,
));
landlock_grants.extend(resolve_staged_runtime_landlock_grants(policy));
landlock_grants.extend(resolve_sealed_runtime_landlock_grants(tier));
landlock_grants.extend(resolve_sealed_etc_landlock_grants(tier));
let sealed_bind_dirs = resolve_sealed_bind_dirs(&exec_allow_paths, policy);
let staged_runtime_dirs = resolve_staged_runtime_dirs(policy);
Ok(Self {
tier,
network_allow_ips,
network_allow_rules,
egress_tcp_ports,
unix_sockets_allowed: policy.unix_sockets_allowed,
workdir_exec: policy.workdir_exec_allowed,
landlock_grants,
sealed_bind_dirs,
staged_runtime_dirs,
resource_limits: policy.resources,
cgroup_scope: None,
workdir_guard: None,
})
}
pub(crate) fn with_host_bounding(
mut self,
cgroup_scope: Option<Arc<crate::cgroup::CgroupScope>>,
workdir_guard: Option<Arc<crate::resources::WorkdirGuard>>,
) -> Self {
self.cgroup_scope = cgroup_scope;
self.workdir_guard = workdir_guard;
self
}
pub(crate) fn nproc_bound(&self) -> crate::resources::NprocBound {
if self
.cgroup_scope
.as_ref()
.is_some_and(|scope| scope.places_subprocesses())
{
crate::resources::NprocBound::CgroupPidsMax
} else {
crate::resources::NprocBound::Headroom {
baseline: crate::resources::uid_task_count().unwrap_or(0),
}
}
}
pub(crate) fn workdir_breach(&self) -> Option<crate::resources::WorkdirBreach> {
self.workdir_guard.as_ref().and_then(|guard| guard.breach())
}
pub(crate) fn check_workdir_budget(&self) -> Result<(), String> {
match self.workdir_breach() {
Some(breach) => Err(crate::resources::workdir_breach_message(breach)),
None => Ok(()),
}
}
#[cfg_attr(not(test), allow(dead_code))]
pub(crate) fn environment_only() -> Self {
Self {
tier: EnforcementTier::EnvironmentOnly,
network_allow_ips: Vec::new(),
network_allow_rules: Vec::new(),
egress_tcp_ports: Vec::new(),
unix_sockets_allowed: false,
workdir_exec: false,
landlock_grants: Vec::new(),
sealed_bind_dirs: Vec::new(),
staged_runtime_dirs: Vec::new(),
resource_limits: crate::resources::HostResourceLimits::default(),
cgroup_scope: None,
workdir_guard: None,
}
}
}
const KERNEL_ENFORCEMENT_NOTICE: &str = "capabilities.shell.allow is non-empty and this host \
resolved to a Linux kernel-enforcement tier (Landlock/seccomp). Landlock now grants a narrow, \
derived read+execute scope outside the workdir (the allowlisted binaries, their loader, and their \
shared libraries — nothing writable, no directory granted wholesale), plus a fixed device set \
every capsule gets unconditionally: /dev/null read+write (the only writable path outside the \
workdir), \
/dev/zero and /dev/urandom read-only, and no other device — /dev/random, /dev/full, /dev/tty and \
raw block devices stay denied. The workdir's own grant also \
withholds device-node and unix-socket creation; seccomp refuses socket(AF_UNIX) outright unless \
capabilities.network.unix_sockets is declared, and always refuses AF_NETLINK/AF_PACKET, so a \
capsule cannot reach a host daemon socket such as /var/run/docker.sock; the seccomp filter now \
defaults to deny and permits only a fixed syscall allowlist modelled on the OCI/Docker default \
profile, so io_uring, bpf, userfaultfd, perf_event_open, ptrace and similar are refused outright; \
and the forked shell child \
drops every Linux capability and sets no_new_privs before execve. Paths outside the workdir are \
denied, not absent: stat, access and readlink on an ungranted path still succeed.";
const KERNEL_SEALED_ENFORCEMENT_NOTICE: &str = "capabilities.shell.allow is non-empty and this \
host resolved to the sealed kernel-enforcement tier: the subprocess tree runs in a private mount \
namespace pivoted onto a composed root (host runtime bind-mounted read-only, the session workdir \
the only writable path, a private /dev tmpfs carrying the OCI default device set), with Landlock \
and seccomp still installed inside it as defence in depth. Paths outside that root are absent \
rather than denied — with one documented exception: /proc is a bind of the host's, not a masked \
private procfs, because mounting one unprivileged needs a PID namespace this tier does not create, \
so host process metadata stays visible there exactly as it is under scoped.";
const SECCOMP_ONLY_WARNING: &str = "capabilities.shell.allow is non-empty and this Linux kernel \
lacks Landlock (kernel <5.13) — filesystem access outside the capsule workdir is not \
kernel-enforced at all, and the seccomp exec/network enforcement that would apply has not been \
verified on real Linux hardware. The fixed capsule device set the Full tier applies (/dev/null \
read+write, /dev/zero and /dev/urandom read-only, every other device denied) is a Landlock \
mechanism, so it does not apply here either: on this tier every device under /dev is reachable \
exactly as it would be without any sandbox. Seccomp also refuses socket(AF_UNIX) unless \
capabilities.network.unix_sockets is declared, and always refuses AF_NETLINK/AF_PACKET — that \
rule needs no Landlock and so applies identically on this tier, but it has not been verified \
either. The same is true of the default-deny syscall allowlist (modelled on the OCI/Docker default \
profile), which refuses io_uring, bpf, userfaultfd, perf_event_open, ptrace and similar on this \
tier too. The forked shell child still drops every Linux capability and \
sets no_new_privs before execve on this tier, independently of Landlock, but that has not been \
verified either. Treat shell subprocess isolation as experimental on this host.";
const ENVIRONMENT_ONLY_WARNING: &str = "capabilities.shell.allow is non-empty but this \
platform has no kernel-level subprocess sandbox (Landlock/seccomp are Linux-only) — \
enforcement is environment-only (synthetic HOME + credential env-stripping). This is \
permanent on this platform.";
pub(crate) fn tier_warning(
tier: EnforcementTier,
shell_allow_is_empty: bool,
) -> Option<(&'static str, &'static str)> {
if shell_allow_is_empty {
return None;
}
match tier {
EnforcementTier::KernelSealed => Some((W_SEC_005, KERNEL_SEALED_ENFORCEMENT_NOTICE)),
EnforcementTier::KernelFull => Some((W_SEC_005, KERNEL_ENFORCEMENT_NOTICE)),
EnforcementTier::KernelSeccompOnly => Some((W_SEC_002, SECCOMP_ONLY_WARNING)),
EnforcementTier::EnvironmentOnly => Some((W_SEC_001, ENVIRONMENT_ONLY_WARNING)),
}
}
const NO_AGGREGATE_BOUNDING_WARNING: &str = "this capsule can spawn native subprocesses, but \
this platform has no cgroup v2 (Linux-only), so no aggregate bound exists across the subprocess \
tree: nothing caps its total memory, task count or CPU. Per-process rlimits from \
capabilities.resources still apply (hard, not soft), but RLIMIT_NPROC is a per-uid ceiling and \
does NOT stop a fork bomb of distinct, short-lived processes — only a cgroup's pids.max does. \
This is permanent on this platform, not a placeholder for a future slice.";
pub(crate) fn aggregate_bounding_warning(
is_linux: bool,
requires_bounding: bool,
has_scope: bool,
) -> Option<(&'static str, &'static str)> {
if is_linux || !requires_bounding || has_scope {
return None;
}
Some((W_SEC_010, NO_AGGREGATE_BOUNDING_WARNING))
}
pub(crate) fn warn_for_missing_aggregate_bounding(
workdir: &Path,
requires_bounding: bool,
has_scope: bool,
) {
let is_linux = cfg!(target_os = "linux");
if let Some((code, message)) =
aggregate_bounding_warning(is_linux, requires_bounding, has_scope)
{
let link = security_warning_link(code);
crate::runtime_err!("[capsule-runtime] warning[{code}]: {message} ({link})");
crate::agent::append_bootstrap_log(
workdir,
&format!("[capability-policy] warning[{code}]: {message} ({link})"),
);
}
}
pub(crate) fn warn_for_enforcement_tier(
tier: EnforcementTier,
workdir: &Path,
policy: &CapabilityPolicy,
) {
if let Some((code, message)) = tier_warning(tier, policy.shell_allow.is_empty()) {
let link = security_warning_link(code);
crate::runtime_err!("[capsule-runtime] warning[{code}]: {message} ({link})");
crate::agent::append_bootstrap_log(
workdir,
&format!("[capability-policy] warning[{code}]: {message} ({link})"),
);
}
if tier == EnforcementTier::EnvironmentOnly {
crate::runtime::warn_if_bash_network_bypass(workdir, policy);
}
}
#[derive(Debug)]
pub(crate) enum SupervisorHandle {
Noop,
#[cfg(target_os = "linux")]
Linux {
proxy_rx: std::sync::mpsc::Receiver<Option<crate::egress_proxy::EgressProxyHandle>>,
diag_read: std::os::fd::OwnedFd,
},
}
impl SupervisorHandle {
pub(crate) fn join_best_effort(self) {
match self {
SupervisorHandle::Noop => {}
#[cfg(target_os = "linux")]
SupervisorHandle::Linux { proxy_rx, .. } => {
if let Ok(Some(proxy)) = proxy_rx.recv_timeout(std::time::Duration::from_secs(5)) {
proxy.shutdown();
}
}
}
}
pub(crate) fn read_diagnostic(&self) -> Option<String> {
match self {
SupervisorHandle::Noop => None,
#[cfg(target_os = "linux")]
SupervisorHandle::Linux { diag_read, .. } => {
use std::os::fd::AsRawFd;
linux_enforce::read_diagnostic_pipe(diag_read.as_raw_fd())
}
}
}
}
#[cfg(test)]
thread_local! {
pub(crate) static FORCE_PREPARE_FAILURE: std::cell::Cell<bool> =
const { std::cell::Cell::new(false) };
}
#[cfg(test)]
fn forced_prepare_failure() -> Result<(), String> {
if FORCE_PREPARE_FAILURE.with(|flag| flag.get()) {
return Err("sandbox: kernel enforcement setup failed (forced by test seam)".to_string());
}
Ok(())
}
#[cfg(all(test, target_os = "linux"))]
thread_local! {
pub(crate) static FORCE_NO_NEW_PRIVS_FAILURE: std::cell::Cell<bool> =
const { std::cell::Cell::new(false) };
pub(crate) static FORCE_LANDLOCK_FAILURE: std::cell::Cell<bool> =
const { std::cell::Cell::new(false) };
pub(crate) static FORCE_SEALED_ROOT_FAILURE: std::cell::Cell<bool> =
const { std::cell::Cell::new(false) };
}
#[cfg(all(test, target_os = "linux"))]
pub(crate) struct ForceSealedRootFailureGuard;
#[cfg(all(test, target_os = "linux"))]
impl ForceSealedRootFailureGuard {
pub(crate) fn new() -> Self {
FORCE_SEALED_ROOT_FAILURE.with(|flag| flag.set(true));
Self
}
}
#[cfg(all(test, target_os = "linux"))]
impl Drop for ForceSealedRootFailureGuard {
fn drop(&mut self) {
FORCE_SEALED_ROOT_FAILURE.with(|flag| flag.set(false));
}
}
#[cfg(all(test, target_os = "linux"))]
pub(crate) fn sealed_test_enforcement() -> ShellEnforcement {
ShellEnforcement {
tier: EnforcementTier::KernelSealed,
..ShellEnforcement::environment_only()
}
}
#[cfg(unix)]
pub(crate) fn attach_process_limits(
command: &mut std::process::Command,
enforcement: &ShellEnforcement,
) {
use std::os::unix::process::CommandExt;
let limits = enforcement.resource_limits;
let nproc = enforcement.nproc_bound();
let cgroup_scope = enforcement.cgroup_scope.clone();
#[allow(unsafe_code)]
unsafe {
command.pre_exec(move || {
crate::resources::apply_hard_rlimits(&limits, nproc)?;
if let Some(scope) = cgroup_scope.as_ref() {
scope.join_current_process()?;
}
Ok(())
});
}
}
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(crate) const FD_HYGIENE_FIRST_FD: u32 = 3;
#[cfg(not(target_os = "linux"))]
pub(crate) fn apply_fd_hygiene(_command: &mut std::process::Command) {}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
pub(crate) fn apply_fd_hygiene(command: &mut std::process::Command) {
use std::os::unix::process::CommandExt;
unsafe {
command.pre_exec(linux_enforce::mark_inherited_fds_cloexec);
}
}
#[cfg(not(target_os = "linux"))]
pub(crate) fn prepare_enforcement(
command: &mut std::process::Command,
enforcement: &ShellEnforcement,
_accessible_workdir: &Path,
_session_workdir: &Path,
) -> Result<SupervisorHandle, String> {
#[cfg(test)]
forced_prepare_failure()?;
debug_assert_eq!(
enforcement.tier,
EnforcementTier::EnvironmentOnly,
"non-Linux targets must always resolve to EnforcementTier::EnvironmentOnly"
);
attach_process_limits(command, enforcement);
Ok(SupervisorHandle::Noop)
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
pub(crate) fn prepare_enforcement(
command: &mut std::process::Command,
enforcement: &ShellEnforcement,
accessible_workdir: &Path,
session_workdir: &Path,
) -> Result<SupervisorHandle, String> {
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
use std::os::unix::process::CommandExt;
#[cfg(test)]
forced_prepare_failure()?;
if enforcement.tier == EnforcementTier::EnvironmentOnly {
attach_process_limits(command, enforcement);
return Ok(SupervisorHandle::Noop);
}
let landlock_fds = if matches!(
enforcement.tier,
EnforcementTier::KernelFull | EnforcementTier::KernelSealed
) {
Some(linux_enforce::open_landlock_fds(
accessible_workdir,
session_workdir,
&enforcement.landlock_grants,
&landlock_device_grants(enforcement.tier),
enforcement.workdir_exec,
enforcement.tier,
)?)
} else {
None
};
let sealed_spec = if enforcement.tier == EnforcementTier::KernelSealed {
Some(build_sealed_root(
accessible_workdir,
session_workdir,
&enforcement.sealed_bind_dirs,
&enforcement.staged_runtime_dirs,
)?)
} else {
None
};
let mut fds = [0i32; 2];
let rc = unsafe {
libc::socketpair(
libc::AF_UNIX,
libc::SOCK_DGRAM | libc::SOCK_CLOEXEC,
0,
fds.as_mut_ptr(),
)
};
if rc != 0 {
return Err(format!(
"sandbox: socketpair() failed: {}",
std::io::Error::last_os_error()
));
}
let (parent_sock, child_sock) =
unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) };
let mut diag_fds = [0i32; 2];
let rc = unsafe { libc::pipe2(diag_fds.as_mut_ptr(), libc::O_CLOEXEC) };
if rc != 0 {
return Err(format!(
"sandbox: pipe2() for enforcement diagnostics failed: {}",
std::io::Error::last_os_error()
));
}
let (diag_read, diag_write) = unsafe {
(
OwnedFd::from_raw_fd(diag_fds[0]),
OwnedFd::from_raw_fd(diag_fds[1]),
)
};
let netns_plan =
crate::network_namespace::CapsuleNetnsPlan::resolve(enforcement.egress_tcp_ports.clone());
let expected_namespace_sockets = netns_plan.socket_count();
let egress_policy = crate::egress_proxy::EgressPolicy::new(
enforcement.network_allow_rules.clone(),
enforcement.network_allow_ips.clone(),
);
let tier = enforcement.tier;
let unix_sockets_allowed = enforcement.unix_sockets_allowed;
let resource_limits = enforcement.resource_limits;
let nproc = enforcement.nproc_bound();
let cgroup_scope = enforcement.cgroup_scope.clone();
unsafe {
command.pre_exec(move || {
if let Err(error) = crate::resources::apply_hard_rlimits(&resource_limits, nproc) {
linux_enforce::write_diagnostic(diag_write.as_raw_fd(), &error.to_string());
return Err(error);
}
if let Some(scope) = cgroup_scope.as_ref() {
if let Err(error) = scope.join_current_process() {
linux_enforce::write_diagnostic(diag_write.as_raw_fd(), &error.to_string());
return Err(error);
}
}
let sock_fd = child_sock.as_raw_fd();
match linux_enforce::child_install_enforcement(
tier,
&netns_plan,
sealed_spec.as_ref(),
landlock_fds.as_ref(),
sock_fd,
unix_sockets_allowed,
) {
Ok(()) => Ok(()),
Err(error) => {
linux_enforce::write_diagnostic(diag_write.as_raw_fd(), &error.to_string());
Err(error)
}
}
});
}
Ok(linux_enforce::start_supervisor(
parent_sock,
egress_policy,
expected_namespace_sockets,
diag_read,
))
}
#[cfg(target_os = "linux")]
fn landlock_device_grants(tier: EnforcementTier) -> Vec<CapsuleDeviceGrant> {
match tier {
EnforcementTier::KernelSealed => crate::sealed::SEALED_DEVICE_NODES
.iter()
.map(|device| CapsuleDeviceGrant {
path: device.path,
writable: device.writable,
})
.collect(),
_ => CAPSULE_DEVICE_GRANTS.to_vec(),
}
}
#[cfg(target_os = "linux")]
fn ensure_sealed_tmp_store(session_workdir: &Path) -> Result<PathBuf, String> {
let tmp_store = session_workdir.join(crate::sealed::SEALED_TMP_DIR_NAME);
std::fs::create_dir_all(&tmp_store).map_err(|error| {
format!(
"sealed: failed to create the workdir-backed /tmp store at {}: {error}",
tmp_store.display()
)
})?;
Ok(tmp_store)
}
#[cfg(target_os = "linux")]
fn ensure_sealed_identity_files(session_workdir: &Path) -> Result<Vec<PathBuf>, String> {
use std::os::unix::fs::MetadataExt;
let staging = session_workdir.join(crate::sealed::SEALED_ETC_STAGING_DIR_NAME);
std::fs::create_dir_all(&staging).map_err(|error| {
format!(
"sealed: failed to create the synthetic /etc staging directory at {}: {error}",
staging.display()
)
})?;
let owner = std::fs::metadata(session_workdir).map_err(|error| {
format!(
"sealed: failed to stat the session workdir {} for the capsule's uid/gid: {error}",
session_workdir.display()
)
})?;
let home = session_workdir.join(crate::shell::SYNTHETIC_HOME_DIR_NAME);
let home = home.to_string_lossy().into_owned();
let identity = crate::sealed::SealedAccountIdentity {
uid: owner.uid(),
gid: owner.gid(),
home: &home,
};
let mut written = Vec::new();
for entry in crate::sealed::SEALED_ETC_PATHS {
let Some(file) = entry.synthetic else {
continue;
};
let path = crate::sealed::synthetic_etc_source(session_workdir, file);
std::fs::write(&path, file.render(&identity)?).map_err(|error| {
format!(
"sealed: failed to write the synthetic {} at {}: {error}",
entry.path,
path.display()
)
})?;
written.push(path);
}
Ok(written)
}
#[cfg(target_os = "linux")]
fn build_sealed_root(
accessible_workdir: &Path,
session_workdir: &Path,
extra_read_only: &[PathBuf],
staged_runtime_read_only: &[PathBuf],
) -> Result<crate::sealed::SealedRootSpec, String> {
use crate::sealed;
if !accessible_workdir.is_absolute() {
return Err(format!(
"sealed: capsule workdir {} is not absolute; a composed root reproduces the workdir at \
its own absolute path",
accessible_workdir.display()
));
}
let base = sealed::choose_root_base(
accessible_workdir,
sealed::SEALED_ROOT_BASE_CANDIDATES,
|path| path.is_dir(),
)
.ok_or_else(|| {
format!(
"sealed: no usable base directory for the composed root; none of {:?} exists outside \
the session workdir's own path",
sealed::SEALED_ROOT_BASE_CANDIDATES
)
})?;
ensure_sealed_tmp_store(session_workdir)?;
ensure_sealed_identity_files(session_workdir)?;
let plan = sealed::plan_composed_root(
accessible_workdir,
session_workdir,
&base,
extra_read_only,
staged_runtime_read_only,
&sealed::RealHostLayout,
);
sealed::build_sealed_root_spec(&plan)
}
#[cfg_attr(target_os = "linux", allow(unsafe_code))]
#[cfg(target_os = "linux")]
mod linux_enforce {
use std::io;
use std::os::fd::{AsFd, AsRawFd, OwnedFd, RawFd};
use std::path::Path;
use landlock::{make_bitflags, AccessFs, BitFlags};
use super::{EnforcementTier, LandlockGrant, SupervisorHandle};
const MAX_DIAG_LEN: usize = 1024;
pub(super) struct OpenLandlockGrant {
fd: OwnedFd,
list_dir: bool,
executable: bool,
}
pub(super) struct OpenDeviceGrant {
fd: OwnedFd,
writable: bool,
}
pub(super) struct LandlockChildFds {
workdir_fd: OwnedFd,
workdir_exec: bool,
sealed_tmp_fd: Option<OwnedFd>,
sealed_identity_fds: Vec<OwnedFd>,
grants: Vec<OpenLandlockGrant>,
devices: Vec<OpenDeviceGrant>,
}
impl LandlockChildFds {
#[cfg(test)]
pub(super) fn sealed_tmp_fd(&self) -> Option<&OwnedFd> {
self.sealed_tmp_fd.as_ref()
}
#[cfg(test)]
pub(super) fn sealed_identity_fd_count(&self) -> usize {
self.sealed_identity_fds.len()
}
}
pub(super) const WORKDIR_ACCESS_RIGHTS_NO_EXEC: BitFlags<AccessFs> = make_bitflags!(AccessFs::{
WriteFile
| ReadFile
| ReadDir
| RemoveDir
| RemoveFile
| MakeDir
| MakeReg
| MakeFifo
| MakeSym
});
pub(super) fn workdir_access_rights(workdir_exec: bool) -> BitFlags<AccessFs> {
if workdir_exec {
WORKDIR_ACCESS_RIGHTS_NO_EXEC | AccessFs::Execute
} else {
WORKDIR_ACCESS_RIGHTS_NO_EXEC
}
}
pub(super) fn probe_landlock_full_access() -> Option<bool> {
use landlock::{
Access, AccessFs, CompatLevel, Compatible, PathBeneath, PathFd, Ruleset, RulesetAttr,
RulesetCreatedAttr, RulesetStatus, ABI,
};
const PROBE_FULLY_ENFORCED: i32 = 0;
const PROBE_PARTIAL: i32 = 1;
const PROBE_FAILED: i32 = 2;
let abi = ABI::V1;
let access_all = AccessFs::from_all(abi);
let root_fd = PathFd::new("/").ok()?;
let ruleset = Ruleset::default()
.set_compatibility(CompatLevel::BestEffort)
.handle_access(access_all)
.ok()?
.create()
.ok()?
.add_rule(PathBeneath::new(root_fd, access_all))
.ok()?;
let pid = unsafe { libc::fork() };
if pid < 0 {
return None;
}
if pid == 0 {
let code = match ruleset.restrict_self() {
Ok(status) => {
if matches!(status.ruleset, RulesetStatus::FullyEnforced) {
PROBE_FULLY_ENFORCED
} else {
PROBE_PARTIAL
}
}
Err(_) => PROBE_FAILED,
};
unsafe { libc::_exit(code) }
}
let mut wait_status: libc::c_int = 0;
let waited = unsafe { libc::waitpid(pid, &mut wait_status, 0) };
if waited < 0 || !libc::WIFEXITED(wait_status) {
return None;
}
match libc::WEXITSTATUS(wait_status) {
PROBE_FULLY_ENFORCED => Some(true),
PROBE_PARTIAL => Some(false),
_ => None,
}
}
pub(super) fn child_install_enforcement(
tier: EnforcementTier,
netns_plan: &crate::network_namespace::CapsuleNetnsPlan,
sealed_spec: Option<&crate::sealed::SealedRootSpec>,
landlock_fds: Option<&LandlockChildFds>,
child_sock_fd: RawFd,
unix_sockets_allowed: bool,
) -> io::Result<()> {
unsafe { crate::network_namespace::create_capsule_netns(netns_plan, child_sock_fd)? };
if tier == EnforcementTier::KernelSealed {
if let Some(spec) = sealed_spec {
#[cfg(test)]
if super::FORCE_SEALED_ROOT_FAILURE.with(|flag| flag.get()) {
return Err(io::Error::other(format!(
"{} unshare(CLONE_NEWUSER|CLONE_NEWNS) failed (forced by test seam)",
crate::sealed::SEALED_ROOT_FAILURE_PREFIX
)));
}
crate::sealed::construct_composed_root(spec)
.map_err(|failure| io::Error::other(spec.describe(failure)))?;
}
}
mark_inherited_fds_cloexec()?;
drop_all_capabilities()?;
install_seccomp_filter(unix_sockets_allowed)?;
if matches!(
tier,
EnforcementTier::KernelFull | EnforcementTier::KernelSealed
) {
if let Some(fds) = landlock_fds {
apply_landlock_scope(fds).map_err(io::Error::other)?;
}
}
Ok(())
}
pub(super) fn mark_inherited_fds_cloexec() -> io::Result<()> {
let rc = unsafe {
libc::syscall(
libc::SYS_close_range,
super::FD_HYGIENE_FIRST_FD,
libc::c_uint::MAX,
libc::CLOSE_RANGE_CLOEXEC,
)
};
if rc != 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
const LINUX_CAPABILITY_VERSION_3: u32 = 0x2008_0522;
const CAP_LAST_CAP_FALLBACK: u32 = 63;
#[repr(C)]
#[allow(dead_code)]
struct CapUserHeader {
version: u32,
pid: libc::c_int,
}
#[repr(C)]
#[derive(Clone, Copy)]
#[allow(dead_code)]
struct CapUserData {
effective: u32,
permitted: u32,
inheritable: u32,
}
fn drop_all_capabilities() -> io::Result<()> {
drop_capability_bounding_set()?;
clear_capability_sets()?;
set_no_new_privs()
}
#[allow(unsafe_code)]
fn drop_capability_bounding_set() -> io::Result<()> {
let zero: libc::c_ulong = 0;
for cap in 0..=kernel_last_cap() {
let cap_arg = libc::c_ulong::from(cap);
let rc = unsafe { libc::prctl(libc::PR_CAPBSET_DROP, cap_arg, zero, zero, zero) };
if rc == 0 {
continue;
}
let error = io::Error::last_os_error();
match error.raw_os_error() {
Some(libc::EINVAL) | Some(libc::EPERM) => continue,
_ => return Err(error),
}
}
Ok(())
}
#[allow(unsafe_code)]
fn clear_capability_sets() -> io::Result<()> {
let header = CapUserHeader {
version: LINUX_CAPABILITY_VERSION_3,
pid: 0,
};
let data = [CapUserData {
effective: 0,
permitted: 0,
inheritable: 0,
}; 2];
let rc = unsafe {
libc::syscall(
libc::SYS_capset,
&header as *const CapUserHeader,
data.as_ptr(),
)
};
if rc != 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
#[allow(unsafe_code)]
fn set_no_new_privs() -> io::Result<()> {
#[cfg(test)]
if super::FORCE_NO_NEW_PRIVS_FAILURE.with(|flag| flag.get()) {
return Err(io::Error::other(
"sandbox: no_new_privs (prctl PR_SET_NO_NEW_PRIVS, 1) failed (forced by test seam)",
));
}
let enable: libc::c_ulong = 1;
let zero: libc::c_ulong = 0;
let rc = unsafe { libc::prctl(libc::PR_SET_NO_NEW_PRIVS, enable, zero, zero, zero) };
if rc != 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
#[allow(unsafe_code)]
fn kernel_last_cap() -> u32 {
const PATH: &[u8] = b"/proc/sys/kernel/cap_last_cap\0";
let fd = unsafe {
libc::open(
PATH.as_ptr() as *const libc::c_char,
libc::O_RDONLY | libc::O_CLOEXEC,
)
};
if fd < 0 {
return CAP_LAST_CAP_FALLBACK;
}
let mut buf = [0u8; 16];
let read = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len()) };
unsafe { libc::close(fd) };
if read <= 0 {
return CAP_LAST_CAP_FALLBACK;
}
let parsed = std::str::from_utf8(&buf[..read as usize])
.ok()
.and_then(|text| text.trim().parse::<u32>().ok());
match parsed {
Some(last) if last <= CAP_LAST_CAP_FALLBACK => last,
_ => CAP_LAST_CAP_FALLBACK,
}
}
const LANDLOCK_SYSCALLS: [(&str, i32); 3] = [
("landlock_create_ruleset", 444),
("landlock_add_rule", 445),
("landlock_restrict_self", 446),
];
fn build_seccomp_filter(
unix_sockets_allowed: bool,
) -> io::Result<libseccomp::ScmpFilterContext> {
let mut filter =
libseccomp::ScmpFilterContext::new(libseccomp::ScmpAction::Errno(libc::EPERM))
.map_err(to_io_err)?;
let _ = filter.set_ctl_log(true);
let socket_syscall = libseccomp::ScmpSyscall::from_name("socket").map_err(to_io_err)?;
for domain in super::denied_socket_domains(unix_sockets_allowed) {
filter
.add_rule_conditional(
libseccomp::ScmpAction::Errno(libc::EACCES),
socket_syscall,
&[libseccomp::ScmpArgCompare::new(
0,
libseccomp::ScmpCompareOp::Equal,
domain as u64,
)],
)
.map_err(to_io_err)?;
}
for domain in super::allowed_socket_domains(unix_sockets_allowed) {
filter
.add_rule_conditional(
libseccomp::ScmpAction::Allow,
socket_syscall,
&[libseccomp::ScmpArgCompare::new(
0,
libseccomp::ScmpCompareOp::Equal,
domain as u64,
)],
)
.map_err(to_io_err)?;
}
for name in super::SECCOMP_SYSCALL_ALLOWLIST {
if let Ok(syscall) = libseccomp::ScmpSyscall::from_name(name) {
let _ = filter.add_rule(libseccomp::ScmpAction::Allow, syscall);
}
}
for (name, fallback_nr) in LANDLOCK_SYSCALLS {
let syscall = libseccomp::ScmpSyscall::from_name(name)
.unwrap_or_else(|_| libseccomp::ScmpSyscall::from(fallback_nr));
filter
.add_rule(libseccomp::ScmpAction::Allow, syscall)
.map_err(to_io_err)?;
}
Ok(filter)
}
fn install_seccomp_filter(unix_sockets_allowed: bool) -> io::Result<()> {
build_seccomp_filter(unix_sockets_allowed)?
.load()
.map_err(to_io_err)
}
fn to_io_err<E: std::fmt::Display>(error: E) -> io::Error {
io::Error::other(error.to_string())
}
#[cfg(test)]
const SOCK_FILTER_BYTES: usize = 8;
#[cfg(test)]
fn export_to_memfd<F>(export: F) -> io::Result<Vec<u8>>
where
F: FnOnce(&OwnedFd) -> Result<(), libseccomp::error::SeccompError>,
{
use std::io::{Read, Seek, SeekFrom};
use std::os::fd::FromRawFd;
let raw = unsafe { libc::memfd_create(c"murmur-seccomp-export".as_ptr(), 0) };
if raw < 0 {
return Err(io::Error::last_os_error());
}
let memfd = unsafe { OwnedFd::from_raw_fd(raw) };
export(&memfd).map_err(to_io_err)?;
let mut file = std::fs::File::from(memfd);
file.seek(SeekFrom::Start(0))?;
let mut exported = Vec::new();
file.read_to_end(&mut exported)?;
Ok(exported)
}
#[cfg(test)]
fn seccomp_filter_instruction_count(unix_sockets_allowed: bool) -> io::Result<usize> {
let filter = build_seccomp_filter(unix_sockets_allowed)?;
let exported = export_to_memfd(|fd| filter.export_bpf(fd))?;
Ok(exported.len() / SOCK_FILTER_BYTES)
}
#[cfg(test)]
fn seccomp_filter_pfc(unix_sockets_allowed: bool) -> io::Result<String> {
let filter = build_seccomp_filter(unix_sockets_allowed)?;
let exported = export_to_memfd(|fd| filter.export_pfc(fd))?;
String::from_utf8(exported).map_err(to_io_err)
}
fn apply_landlock_scope(fds: &LandlockChildFds) -> Result<(), String> {
#[cfg(test)]
if super::FORCE_LANDLOCK_FAILURE.with(|flag| flag.get()) {
return Err("landlock: ruleset construction failed (forced by test seam)".to_string());
}
use landlock::{
Access, AccessFs, CompatLevel, Compatible, PathBeneath, Ruleset, RulesetAttr,
RulesetCreatedAttr, ABI,
};
let abi = ABI::V1;
let access_all = AccessFs::from_all(abi);
let read_execute = AccessFs::Execute | AccessFs::ReadFile;
let read_execute_list = read_execute | AccessFs::ReadDir;
let read_list = AccessFs::ReadFile | AccessFs::ReadDir;
let read_only: BitFlags<AccessFs> = AccessFs::ReadFile.into();
let device_read: BitFlags<AccessFs> = AccessFs::ReadFile.into();
let device_read_write = device_read | AccessFs::WriteFile;
let writable_rights = workdir_access_rights(fds.workdir_exec);
let mut ruleset = Ruleset::default()
.set_compatibility(CompatLevel::HardRequirement)
.handle_access(access_all)
.map_err(|error| format!("landlock: handle_access failed: {error}"))?
.create()
.map_err(|error| format!("landlock: ruleset create failed: {error}"))?
.add_rule(PathBeneath::new(fds.workdir_fd.as_fd(), writable_rights))
.map_err(|error| format!("landlock: add_rule failed: {error}"))?;
if let Some(sealed_tmp_fd) = &fds.sealed_tmp_fd {
ruleset = ruleset
.add_rule(PathBeneath::new(sealed_tmp_fd.as_fd(), writable_rights))
.map_err(|error| {
format!("landlock: add_rule for the workdir-backed /tmp failed: {error}")
})?;
}
for identity_fd in &fds.sealed_identity_fds {
ruleset = ruleset
.add_rule(PathBeneath::new(identity_fd.as_fd(), read_only))
.map_err(|error| {
format!("landlock: add_rule for a synthetic /etc file failed: {error}")
})?;
}
for grant in &fds.grants {
let access = match (grant.executable, grant.list_dir) {
(true, true) => read_execute_list,
(true, false) => read_execute,
(false, true) => read_list,
(false, false) => read_only,
};
ruleset = ruleset
.add_rule(PathBeneath::new(grant.fd.as_fd(), access))
.map_err(|error| format!("landlock: add_rule for grant failed: {error}"))?;
}
for device in &fds.devices {
let access = if device.writable {
device_read_write
} else {
device_read
};
ruleset = ruleset
.add_rule(PathBeneath::new(device.fd.as_fd(), access))
.map_err(|error| format!("landlock: add_rule for device failed: {error}"))?;
}
ruleset
.restrict_self()
.map_err(|error| format!("landlock: restrict_self failed: {error}"))?;
Ok(())
}
pub(super) fn open_landlock_fds(
accessible_workdir: &Path,
session_workdir: &Path,
landlock_grants: &[LandlockGrant],
device_grants: &[super::CapsuleDeviceGrant],
workdir_exec: bool,
tier: EnforcementTier,
) -> Result<LandlockChildFds, String> {
let workdir_fd = open_o_path(accessible_workdir)
.map(OwnedFd::from)
.map_err(|error| {
format!(
"sandbox: failed to open workdir {} for Landlock scoping: {error}",
accessible_workdir.display()
)
})?;
let sealed_tmp_fd = if tier == EnforcementTier::KernelSealed {
let tmp_store = super::ensure_sealed_tmp_store(session_workdir)?;
let tmp_store_file = open_o_path(&tmp_store).map_err(|error| {
format!(
"sandbox: failed to open the workdir-backed /tmp store {} for Landlock \
scoping: {error}",
tmp_store.display()
)
})?;
Some(OwnedFd::from(tmp_store_file))
} else {
None
};
let sealed_identity_fds = if tier == EnforcementTier::KernelSealed {
let mut fds = Vec::new();
for path in super::ensure_sealed_identity_files(session_workdir)? {
let file = open_o_path(&path).map_err(|error| {
format!(
"sandbox: failed to open the synthetic /etc file {} for Landlock scoping: \
{error}",
path.display()
)
})?;
fds.push(OwnedFd::from(file));
}
fds
} else {
Vec::new()
};
let mut grants = Vec::with_capacity(landlock_grants.len());
for grant in landlock_grants {
match open_o_path(&grant.path) {
Ok(file) => grants.push(OpenLandlockGrant {
list_dir: grant.list_dir
&& file.metadata().map(|meta| meta.is_dir()).unwrap_or(false),
fd: OwnedFd::from(file),
executable: grant.executable,
}),
Err(_) => continue,
}
}
let mut devices = Vec::with_capacity(device_grants.len());
for device in device_grants {
match open_o_path(Path::new(device.path)) {
Ok(file) => devices.push(OpenDeviceGrant {
fd: OwnedFd::from(file),
writable: device.writable,
}),
Err(_) => continue,
}
}
Ok(LandlockChildFds {
workdir_fd,
workdir_exec,
sealed_tmp_fd,
sealed_identity_fds,
grants,
devices,
})
}
fn open_o_path(path: &Path) -> io::Result<std::fs::File> {
use std::os::unix::fs::OpenOptionsExt;
std::fs::OpenOptions::new()
.read(true)
.custom_flags(libc::O_PATH | libc::O_CLOEXEC)
.open(path)
}
#[allow(unsafe_code)]
pub(super) fn write_diagnostic(fd: RawFd, message: &str) {
let bytes = message.as_bytes();
let len = bytes.len().min(MAX_DIAG_LEN);
let mut written = 0;
while written < len {
let n = unsafe {
libc::write(
fd,
bytes[written..len].as_ptr() as *const libc::c_void,
len - written,
)
};
if n < 0 {
if io::Error::last_os_error().raw_os_error() == Some(libc::EINTR) {
continue;
}
break;
}
if n == 0 {
break;
}
written += n as usize;
}
}
#[allow(unsafe_code)]
pub(super) fn read_diagnostic_pipe(fd: RawFd) -> Option<String> {
let mut buf = Vec::new();
let mut chunk = [0u8; 256];
loop {
let n = unsafe { libc::read(fd, chunk.as_mut_ptr() as *mut libc::c_void, chunk.len()) };
if n < 0 {
if io::Error::last_os_error().raw_os_error() == Some(libc::EINTR) {
continue;
}
break;
}
if n == 0 {
break;
}
buf.extend_from_slice(&chunk[..n as usize]);
if buf.len() >= MAX_DIAG_LEN {
break;
}
}
if buf.is_empty() {
None
} else {
Some(String::from_utf8_lossy(&buf).into_owned())
}
}
pub(super) fn start_supervisor(
parent_sock: OwnedFd,
egress_policy: crate::egress_proxy::EgressPolicy,
expected_namespace_sockets: usize,
diag_read: OwnedFd,
) -> SupervisorHandle {
let (proxy_tx, proxy_rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let proxy = start_namespace_proxy(
parent_sock.as_raw_fd(),
expected_namespace_sockets,
egress_policy,
);
if let Err(returned) = proxy_tx.send(proxy) {
if let Some(proxy) = returned.0 {
proxy.shutdown();
}
}
});
SupervisorHandle::Linux {
proxy_rx,
diag_read,
}
}
fn start_namespace_proxy(
sock_fd: RawFd,
expected: usize,
policy: crate::egress_proxy::EgressPolicy,
) -> Option<crate::egress_proxy::EgressProxyHandle> {
let sockets = match crate::network_namespace::receive_namespace_sockets(sock_fd, expected) {
Ok(sockets) => sockets,
Err(error) => {
crate::runtime_err!(
"[capsule-runtime] warning: failed to receive the capsule's network \
namespace sockets: {error} (the subprocess spawn reports the underlying \
pre_exec failure independently)"
);
return None;
}
};
match crate::egress_proxy::start_egress_proxy(sockets, policy) {
Ok(proxy) => Some(proxy),
Err(error) => {
crate::runtime_err!(
"[capsule-runtime] warning: failed to serve the capsule's network namespace \
sockets: {error} (the subprocess tree has no route off this host, so its \
network calls fail closed)"
);
None
}
}
}
#[cfg(test)]
#[allow(clippy::print_stdout, clippy::print_stderr)]
mod seccomp_budget {
use std::io;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
use super::{
build_seccomp_filter, export_to_memfd, seccomp_filter_pfc, to_io_err,
LANDLOCK_SYSCALLS, SOCK_FILTER_BYTES,
};
use crate::sandbox::{
allowed_socket_domains, denied_socket_domains, detect_enforcement_tier,
EnforcementTier, BPF_MAXINSNS_CEILING, SECCOMP_FILTER_INSTRUCTION_BUDGET,
SECCOMP_MAX_INSNS_PER_PATH, SECCOMP_SYSCALL_ALLOWLIST,
};
const KERNEL_PER_FILTER_OVERHEAD_INSNS: usize = 4;
const PROBE_ERRNO_UNKNOWN: i32 = 254;
fn probe_filter(rules: usize) -> io::Result<(libseccomp::ScmpFilterContext, usize)> {
let mut filter = libseccomp::ScmpFilterContext::new(libseccomp::ScmpAction::Allow)
.map_err(to_io_err)?;
filter.set_api_sysrawrc(true).map_err(to_io_err)?;
let syscall = libseccomp::ScmpSyscall::from_name("ptrace").map_err(to_io_err)?;
for value in 0..rules {
filter
.add_rule_conditional(
libseccomp::ScmpAction::Errno(libc::EPERM),
syscall,
&[libseccomp::ScmpArgCompare::new(
0,
libseccomp::ScmpCompareOp::Equal,
value as u64,
)],
)
.map_err(to_io_err)?;
}
let exported = export_to_memfd(|fd| filter.export_bpf(fd))?;
Ok((filter, exported.len() / SOCK_FILTER_BYTES))
}
fn first_rule_count_above(instructions: usize) -> io::Result<usize> {
let mut under = 1;
let mut over = 8;
while probe_filter(over)?.1 <= instructions {
under = over;
over *= 2;
assert!(
over <= 1 << 20,
"no probe filter of up to {over} rules exceeded {instructions} instructions — \
the probe is not growing the program the way this search assumes"
);
}
while over - under > 1 {
let mid = under + (over - under) / 2;
if probe_filter(mid)?.1 <= instructions {
under = mid;
} else {
over = mid;
}
}
Ok(over)
}
fn refusal_errno(error: &libseccomp::error::SeccompError) -> i32 {
use libseccomp::error::SeccompErrno;
if let Some(raw) = error.sysrawrc() {
return -raw;
}
match error.errno() {
Some(SeccompErrno::EACCES) => libc::EACCES,
Some(SeccompErrno::ECANCELED) => libc::ECANCELED,
Some(SeccompErrno::EINVAL) => libc::EINVAL,
Some(SeccompErrno::ENOMEM) => libc::ENOMEM,
_ => PROBE_ERRNO_UNKNOWN,
}
}
fn load_in_child(filter: &libseccomp::ScmpFilterContext) -> Result<(), i32> {
let pid = unsafe { libc::fork() };
assert!(pid >= 0, "fork failed: {}", io::Error::last_os_error());
if pid == 0 {
let code = match filter.load() {
Ok(()) => 0,
Err(error) => refusal_errno(&error),
};
unsafe { libc::_exit(code) }
}
let mut status: libc::c_int = 0;
let waited = unsafe { libc::waitpid(pid, &mut status, 0) };
assert!(waited == pid, "waitpid: {}", io::Error::last_os_error());
assert!(
libc::WIFEXITED(status),
"the load probe child did not exit normally (status {status})"
);
match libc::WEXITSTATUS(status) {
0 => Ok(()),
errno => Err(errno),
}
}
fn resolvable_allowlist_syscalls() -> io::Result<usize> {
use std::collections::HashSet;
let mut scratch =
libseccomp::ScmpFilterContext::new(libseccomp::ScmpAction::Errno(libc::EPERM))
.map_err(to_io_err)?;
let mut accepted = HashSet::new();
for name in SECCOMP_SYSCALL_ALLOWLIST {
if let Ok(syscall) = libseccomp::ScmpSyscall::from_name(name) {
if scratch
.add_rule(libseccomp::ScmpAction::Allow, syscall)
.is_ok()
{
accepted.insert(i32::from(syscall));
}
}
}
Ok(accepted.len())
}
fn skip_without_seccomp(what: &str) -> bool {
let tier = detect_enforcement_tier();
if tier == EnforcementTier::EnvironmentOnly {
eprintln!(
"SKIP — PROVES NOTHING ON THIS HOST: measuring {what} requires a host that \
can load a seccomp filter at all, which is every tier above \
EnforcementTier::EnvironmentOnly; detected {tier:?}"
);
return true;
}
false
}
#[test]
fn seccomp_filter_instruction_count() {
for unix_sockets_allowed in [false, true] {
let count = super::seccomp_filter_instruction_count(unix_sockets_allowed)
.expect("export the built seccomp filter");
eprintln!(
"seccomp filter instruction count: unix_sockets_allowed={unix_sockets_allowed} \
-> {count} instructions; budget {SECCOMP_FILTER_INSTRUCTION_BUDGET}; kernel \
per-filter ceiling {BPF_MAXINSNS_CEILING} ({:.1}x headroom); kernel \
cumulative ceiling {SECCOMP_MAX_INSNS_PER_PATH} ({:.1}x headroom)",
BPF_MAXINSNS_CEILING as f64 / count as f64,
SECCOMP_MAX_INSNS_PER_PATH as f64 / count as f64,
);
assert!(
count > 0,
"the exported filter is empty, so nothing was measured — \
`export_bpf` wrote no `struct sock_filter` records at all"
);
assert!(
count <= SECCOMP_FILTER_INSTRUCTION_BUDGET,
"the child's seccomp filter compiles to {count} instructions with \
unix_sockets_allowed={unix_sockets_allowed}, over this project's budget of \
{SECCOMP_FILTER_INSTRUCTION_BUDGET} (the kernel refuses a single filter over \
{BPF_MAXINSNS_CEILING} with EINVAL). Re-measure, decide whether the rule set \
still earns its size, and raise SECCOMP_FILTER_INSTRUCTION_BUDGET \
deliberately — updating \
docs/content/reference/seccomp-filter-instruction-budget.md with the new \
numbers — rather than letting the filter grow into a spawn failure"
);
}
}
#[test]
fn seccomp_filter_rule_shape() {
let allowlist_syscalls =
resolvable_allowlist_syscalls().expect("count the resolvable allowlist syscalls");
assert!(
allowlist_syscalls > 0,
"no SECCOMP_SYSCALL_ALLOWLIST name resolved on this host, so this test would pass \
for the wrong reason"
);
for unix_sockets_allowed in [false, true] {
let filter =
build_seccomp_filter(unix_sockets_allowed).expect("build the seccomp filter");
assert_eq!(
filter.get_act_default().expect("read the default action"),
libseccomp::ScmpAction::Errno(libc::EPERM),
"the filter's default action is no longer a deny with EPERM, which is what \
makes every syscall outside the allowlist unreachable"
);
let pfc =
seccomp_filter_pfc(unix_sockets_allowed).expect("export the filter's pfc");
let denied = denied_socket_domains(unix_sockets_allowed);
let allowed = allowed_socket_domains(unix_sockets_allowed);
let syscall_blocks = pfc.matches("# filter for syscall ").count();
assert_eq!(
syscall_blocks,
1 + allowlist_syscalls + LANDLOCK_SYSCALLS.len(),
"the built filter carries rules for {syscall_blocks} syscalls, not the \
socket rule plus {allowlist_syscalls} allowlist syscalls plus {} Landlock \
syscalls this code adds",
LANDLOCK_SYSCALLS.len()
);
let deny_actions = pfc
.matches(&format!("action ERRNO({});", libc::EACCES))
.count();
assert_eq!(
deny_actions,
denied.len(),
"the built filter carries {deny_actions} EACCES rules, not one per denied \
socket domain ({denied:?}) — an unconditional Allow on `socket` would look \
exactly like this, and would have discarded every domain denial with it"
);
let allow_actions = pfc.matches("action ALLOW;").count();
assert_eq!(
allow_actions,
allowed.len() + allowlist_syscalls + LANDLOCK_SYSCALLS.len(),
"the built filter carries {allow_actions} Allow rules, not one per allowed \
socket domain ({allowed:?}) plus one per allowlist syscall plus one per \
Landlock syscall"
);
for (name, _) in LANDLOCK_SYSCALLS {
assert!(
pfc.contains(&format!("\"{name}\"")),
"{name} has no rule in the built filter, so `apply_landlock_scope` would \
be refused by the filter installed just before it"
);
}
eprintln!(
"seccomp filter rule shape: unix_sockets_allowed={unix_sockets_allowed} -> \
default ERRNO(EPERM); {} socket EACCES rules; {} socket ALLOW rules; \
{allowlist_syscalls} allowlist syscalls; {} Landlock syscalls",
denied.len(),
allowed.len(),
LANDLOCK_SYSCALLS.len(),
);
}
}
#[test]
fn seccomp_per_filter_ceiling() {
if skip_without_seccomp("the kernel's per-filter instruction ceiling") {
return;
}
let over_rules = first_rule_count_above(BPF_MAXINSNS_CEILING)
.expect("bracket the per-filter ceiling");
let (under_filter, under_count) =
probe_filter(over_rules - 1).expect("build the under-ceiling probe filter");
let (over_filter, over_count) =
probe_filter(over_rules).expect("build the over-ceiling probe filter");
assert!(
under_count <= BPF_MAXINSNS_CEILING && over_count > BPF_MAXINSNS_CEILING,
"the two probe filters ({under_count} and {over_count} instructions) do not \
straddle BPF_MAXINSNS_CEILING ({BPF_MAXINSNS_CEILING}), so this test would prove \
nothing about where the ceiling is"
);
let under_outcome = load_in_child(&under_filter);
let over_outcome = load_in_child(&over_filter);
eprintln!(
"seccomp per-filter ceiling: largest filter that loaded = {under_count} \
instructions; smallest that did not = {over_count} instructions, refused with \
errno {} (EINVAL is {}); recorded BPF_MAXINSNS_CEILING = {BPF_MAXINSNS_CEILING}",
over_outcome.err().unwrap_or(0),
libc::EINVAL,
);
assert_eq!(
under_outcome,
Ok(()),
"a {under_count}-instruction filter was refused, below the \
{BPF_MAXINSNS_CEILING}-instruction ceiling this test expects the kernel to hold"
);
assert_eq!(
over_outcome,
Err(libc::EINVAL),
"a {over_count}-instruction filter was not refused with EINVAL, so \
BPF_MAXINSNS_CEILING ({BPF_MAXINSNS_CEILING}) is not where this kernel puts its \
per-filter limit"
);
}
#[test]
fn seccomp_cumulative_ceiling() {
if skip_without_seccomp("the kernel's cumulative filter-chain ceiling") {
return;
}
let (filter, count) = probe_filter(120).expect("build the cumulative probe filter");
let per_filter_cost = count + KERNEL_PER_FILTER_OVERHEAD_INSNS;
let max_loads = SECCOMP_MAX_INSNS_PER_PATH / per_filter_cost + 16;
let mut fds = [0 as libc::c_int; 2];
let rc = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) };
assert!(rc == 0, "pipe2: {}", io::Error::last_os_error());
let (read_fd, write_fd) =
unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) };
let pid = unsafe { libc::fork() };
assert!(pid >= 0, "fork failed: {}", io::Error::last_os_error());
if pid == 0 {
let mut loads: u32 = 0;
let mut refusal: i32 = 0;
while loads < max_loads as u32 {
match filter.load() {
Ok(()) => loads += 1,
Err(error) => {
refusal = refusal_errno(&error);
break;
}
}
}
let mut report = [0u8; 8];
report[..4].copy_from_slice(&loads.to_ne_bytes());
report[4..].copy_from_slice(&refusal.to_ne_bytes());
unsafe { libc::write(write_fd.as_raw_fd(), report.as_ptr().cast(), report.len()) };
unsafe { libc::_exit(0) }
}
drop(write_fd);
let mut report = [0u8; 8];
std::io::Read::read_exact(&mut std::fs::File::from(read_fd), &mut report)
.expect("read the cumulative probe child's report");
let loads = u32::from_ne_bytes(report[..4].try_into().unwrap()) as usize;
let refusal = i32::from_ne_bytes(report[4..].try_into().unwrap());
let mut status: libc::c_int = 0;
let waited = unsafe { libc::waitpid(pid, &mut status, 0) };
assert!(waited == pid, "waitpid: {}", io::Error::last_os_error());
let stacked = loads * count;
let cumulative_at_refusal = count + loads * per_filter_cost;
let loads_if_counted_classic = SECCOMP_MAX_INSNS_PER_PATH / per_filter_cost;
let shipped = super::seccomp_filter_instruction_count(false)
.expect("measure the shipped filter for comparison");
eprintln!(
"seccomp cumulative ceiling: {loads} loads of a {count}-instruction filter \
succeeded ({stacked} instructions stacked, {cumulative_at_refusal} counted the \
way the kernel counts once the {KERNEL_PER_FILTER_OVERHEAD_INSNS}-instruction \
per-filter overhead is added); load {} was refused with errno {refusal} (ENOMEM \
is {}), against the recorded SECCOMP_MAX_INSNS_PER_PATH of \
{SECCOMP_MAX_INSNS_PER_PATH} — which the exported classic length puts at at most \
{loads_if_counted_classic} loads. The runtime installs ONE filter of {shipped} \
instructions, so reaching this ceiling would take {} of them in a single task.",
loads + 1,
libc::ENOMEM,
SECCOMP_MAX_INSNS_PER_PATH / shipped,
);
assert!(
loads >= 2,
"the kernel refused the filter chain after only {loads} load(s), so this is not \
a cumulative limit and the ENOMEM it returns is not the one being ruled out"
);
assert_eq!(
refusal,
libc::ENOMEM,
"stacking filters past SECCOMP_MAX_INSNS_PER_PATH was refused with errno \
{refusal}, not ENOMEM ({}) — the rule-out in \
docs/content/reference/seccomp-filter-instruction-budget.md rests on ENOMEM \
being what this ceiling returns",
libc::ENOMEM
);
assert!(
loads <= loads_if_counted_classic,
"the kernel took {loads} loads of a {count}-instruction filter — more than the \
{loads_if_counted_classic} that fit under SECCOMP_MAX_INSNS_PER_PATH \
({SECCOMP_MAX_INSNS_PER_PATH}) even before the conversion to eBPF lengthens \
each one, so this kernel's ceiling is higher than the recorded value"
);
assert!(
loads * 2 > loads_if_counted_classic,
"the kernel refused after {loads} loads of a {count}-instruction filter, far \
short of the {loads_if_counted_classic} that SECCOMP_MAX_INSNS_PER_PATH \
({SECCOMP_MAX_INSNS_PER_PATH}) implies — the ceiling being measured is not the \
one recorded"
);
assert!(
shipped * 2 < SECCOMP_MAX_INSNS_PER_PATH,
"the shipped filter ({shipped} instructions) is within a factor of two of the \
cumulative ceiling, which would make the rule-out this test supports untrue"
);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use murmur_artifact::security_warnings::W_SEC_003;
#[test]
fn fd_hygiene_range_starts_above_stdio() {
assert_eq!(
FD_HYGIENE_FIRST_FD, 3,
"fd hygiene must start at 3 — 0/1/2 are the stdio pipes both spawn paths need to \
survive execve"
);
}
#[cfg(target_os = "linux")]
#[allow(unsafe_code)]
#[test]
fn close_range_syscall_reports_failure_as_minus_one_and_errno() {
let rc = unsafe {
libc::syscall(
libc::SYS_close_range,
FD_HYGIENE_FIRST_FD,
libc::c_uint::MAX,
libc::c_uint::MAX,
)
};
assert_eq!(rc, -1, "an invalid flag word must not be accepted");
assert_eq!(
std::io::Error::last_os_error().raw_os_error(),
Some(libc::EINVAL),
"the two statements the production path runs on failure must carry the kernel's errno"
);
}
fn sealed_capable() -> crate::sealed::SealedProbe {
crate::sealed::SealedProbe {
userns_grant: crate::sealed::UsernsGrant::ProfileConfining,
namespace: crate::sealed::NamespaceProbe::Ok,
}
}
fn sealed_incapable() -> crate::sealed::SealedProbe {
crate::sealed::SealedProbe::default()
}
fn unprobed() -> HostProbe {
HostProbe::from_parts(None, crate::sealed::SealedProbe::default())
}
#[test]
fn an_unprobed_host_resolves_to_seccomp_only() {
let tier = unprobed().tier();
if cfg!(target_os = "linux") {
assert_eq!(tier, EnforcementTier::KernelSeccompOnly);
} else {
assert_eq!(tier, EnforcementTier::EnvironmentOnly);
}
assert_ne!(tier, EnforcementTier::KernelSealed);
assert_ne!(tier, EnforcementTier::KernelFull);
}
#[test]
fn an_unprobed_host_names_a_blocker() {
assert!(unprobed().sealed_blocker().is_some());
}
#[test]
fn an_unprobed_host_achieves_less_than_sealed() {
let achieved = crate::containment::achieved_class_for_tier(unprobed().tier());
assert!(achieved < murmur_artifact::ContainmentClass::Sealed);
}
#[test]
fn an_unprobed_host_refuses_a_declared_sealed_floor() {
let probe = unprobed();
let achieved = crate::containment::achieved_class_for_tier(probe.tier());
let error = crate::containment::check_containment_floor(
murmur_artifact::ContainmentClass::Sealed,
achieved,
probe.sealed_blocker(),
false,
)
.expect_err("an unprobed host cannot back a declared sealed floor");
assert!(matches!(
error,
crate::errors::RuntimeError::ContainmentFloorUnmet { .. }
));
}
#[test]
fn tier_from_probe_is_identical_across_every_permitting_userns_grant() {
use crate::sealed::{NamespaceProbe, SealedProbe, UsernsGrant};
for grant in UsernsGrant::ALL {
let probe = SealedProbe {
userns_grant: *grant,
namespace: NamespaceProbe::Ok,
};
let expected = if grant.permits_userns() {
EnforcementTier::KernelSealed
} else {
EnforcementTier::KernelFull
};
assert_eq!(
tier_from_probe(true, Some(true), probe),
expected,
"grant {grant:?} must decide the tier only through permits_userns()"
);
}
}
#[test]
fn tier_from_probe_non_linux_is_always_environment_only() {
for landlock in [None, Some(true), Some(false)] {
for sealed in [sealed_capable(), sealed_incapable()] {
assert_eq!(
tier_from_probe(false, landlock, sealed),
EnforcementTier::EnvironmentOnly
);
}
}
}
#[test]
fn tier_from_probe_linux_fully_enforced_is_kernel_full() {
assert_eq!(
tier_from_probe(true, Some(true), sealed_incapable()),
EnforcementTier::KernelFull
);
}
#[test]
fn tier_from_probe_linux_partially_enforced_is_seccomp_only() {
assert_eq!(
tier_from_probe(true, Some(false), sealed_incapable()),
EnforcementTier::KernelSeccompOnly
);
}
#[test]
fn tier_from_probe_linux_no_probe_result_is_seccomp_only() {
assert_eq!(
tier_from_probe(true, None, sealed_incapable()),
EnforcementTier::KernelSeccompOnly
);
}
#[test]
fn tier_from_probe_reaches_sealed_only_when_every_precondition_holds() {
use crate::sealed::{NamespaceProbe, SealedProbe};
assert_eq!(
tier_from_probe(true, Some(true), sealed_capable()),
EnforcementTier::KernelSealed
);
assert_eq!(
tier_from_probe(
true,
Some(true),
SealedProbe {
userns_grant: crate::sealed::UsernsGrant::Withheld,
namespace: NamespaceProbe::Ok
}
),
EnforcementTier::KernelFull
);
for namespace in [
NamespaceProbe::Denied,
NamespaceProbe::MapDenied,
NamespaceProbe::MountDenied,
NamespaceProbe::Unsupported,
] {
assert_eq!(
tier_from_probe(
true,
Some(true),
SealedProbe {
userns_grant: crate::sealed::UsernsGrant::ProfileConfining,
namespace
}
),
EnforcementTier::KernelFull,
"namespace probe {namespace:?} must fall back to KernelFull, never below it"
);
}
assert_eq!(
tier_from_probe(true, Some(false), sealed_capable()),
EnforcementTier::KernelSeccompOnly
);
assert_eq!(
tier_from_probe(true, None, sealed_capable()),
EnforcementTier::KernelSeccompOnly
);
assert_eq!(
tier_from_probe(false, Some(true), sealed_capable()),
EnforcementTier::EnvironmentOnly
);
}
#[test]
fn applied_tier_never_installs_a_composed_root_for_a_weaker_declaration() {
use murmur_artifact::ContainmentClass;
assert_eq!(
applied_tier(EnforcementTier::KernelSealed, ContainmentClass::Sealed),
EnforcementTier::KernelSealed
);
for declared in [ContainmentClass::Advisory, ContainmentClass::Scoped] {
assert_eq!(
applied_tier(EnforcementTier::KernelSealed, declared),
EnforcementTier::KernelFull,
"declaring {declared} must not opt a capsule into the composed root"
);
}
for host_tier in [
EnforcementTier::KernelFull,
EnforcementTier::KernelSeccompOnly,
EnforcementTier::EnvironmentOnly,
] {
for declared in [
ContainmentClass::Advisory,
ContainmentClass::Scoped,
ContainmentClass::Sealed,
] {
assert_eq!(applied_tier(host_tier, declared), host_tier);
}
}
}
#[test]
fn sealed_bind_dirs_are_whole_directories_outside_the_fixed_runtime_paths() {
use murmur_artifact::{InterpreterRuntimeDir, InterpreterRuntimeGrant};
let policy = CapabilityPolicy {
shell_interpreter_runtime: vec![InterpreterRuntimeGrant {
binary: "python3".to_string(),
dirs: vec![
InterpreterRuntimeDir {
path: "/usr/lib/python3.11".to_string(),
list_dir: true,
},
InterpreterRuntimeDir {
path: "/opt/py/lib".to_string(),
list_dir: true,
},
InterpreterRuntimeDir {
path: "/opt/py/lib".to_string(),
list_dir: false,
},
InterpreterRuntimeDir {
path: "relative/lib".to_string(),
list_dir: false,
},
],
}],
..CapabilityPolicy::default()
};
let exec_allow = vec![
PathBuf::from("/usr/bin/python3"),
PathBuf::from("/opt/toolchain/bin/cc"),
];
assert_eq!(
resolve_sealed_bind_dirs(&exec_allow, &policy),
vec![
PathBuf::from("/opt/toolchain/bin"),
PathBuf::from("/opt/py/lib")
]
);
}
#[test]
fn staged_runtime_dirs_keep_fixed_path_overlaps_and_drop_only_duplicates_and_relatives() {
use murmur_artifact::StagedRuntimeGrant;
let grant = |binary: &str, source: &str| StagedRuntimeGrant {
binary: binary.to_string(),
source_path: source.to_string(),
pin: "pin-1".to_string(),
};
let policy = CapabilityPolicy {
shell_staged_runtime: vec![
grant("python3", "/opt/py"),
grant("perl", "/usr/lib/perl5"),
grant("python3.12", "/opt/py"),
grant("ruby", "relative/ruby"),
],
..CapabilityPolicy::default()
};
assert_eq!(
resolve_staged_runtime_dirs(&policy),
vec![PathBuf::from("/opt/py"), PathBuf::from("/usr/lib/perl5")]
);
}
#[test]
fn every_staged_runtime_dir_also_gets_a_listable_landlock_grant() {
use murmur_artifact::StagedRuntimeGrant;
let policy = CapabilityPolicy {
shell_staged_runtime: vec![StagedRuntimeGrant {
binary: "python3".to_string(),
source_path: "/opt/py".to_string(),
pin: "pin-1".to_string(),
}],
..CapabilityPolicy::default()
};
let dirs = resolve_staged_runtime_dirs(&policy);
let grants = resolve_staged_runtime_landlock_grants(&policy);
assert_eq!(
grants,
vec![LandlockGrant {
path: PathBuf::from("/opt/py"),
list_dir: true,
executable: true,
}],
"a staged tree is walked by the runtime that uses it, so it must be listable",
);
assert_eq!(
grants
.iter()
.map(|grant| grant.path.clone())
.collect::<Vec<_>>(),
dirs,
"the bound set and the granted set must not drift apart",
);
}
#[test]
fn every_sealed_runtime_path_gets_a_listable_non_executable_landlock_grant() {
let grants = resolve_sealed_runtime_landlock_grants(EnforcementTier::KernelSealed);
assert_eq!(
grants,
crate::sealed::SEALED_RUNTIME_PATHS
.iter()
.map(|path| LandlockGrant {
path: PathBuf::from(path),
list_dir: true,
executable: false,
})
.collect::<Vec<_>>(),
"the composed root binds exactly these, so exactly these must be enumerable — \
the bound set and the granted set must not drift apart",
);
assert!(
grants.iter().all(|grant| grant.list_dir),
"a runtime tree an interpreter walks is useless without getdents64",
);
assert!(
grants.iter().all(|grant| !grant.executable),
"Execute is the exec allowlist: granting it over /usr, /bin and /sbin would let a \
sealed capsule run every binary the host ships, whatever shell.allow says",
);
}
#[test]
fn sealed_runtime_landlock_grants_are_sealed_tier_only() {
for tier in [
EnforcementTier::KernelFull,
EnforcementTier::KernelSeccompOnly,
EnforcementTier::EnvironmentOnly,
] {
assert!(
resolve_sealed_runtime_landlock_grants(tier).is_empty(),
"{tier:?} has no composed root, so granting ReadDir on SEALED_RUNTIME_PATHS \
would enumerate the real host filesystem",
);
}
}
#[test]
fn shell_enforcement_grants_sealed_runtime_paths_only_when_sealed_applies() {
let policy = CapabilityPolicy::default();
let usr = PathBuf::from("/usr");
let scoped = ShellEnforcement::resolve(
&policy,
murmur_artifact::ContainmentClass::Scoped,
HostProbe::probe(),
)
.expect("an empty policy resolves");
assert!(
!scoped.landlock_grants.iter().any(|grant| grant.path == usr),
"a scoped capsule runs Landlock over the real host filesystem — /usr must stay \
unenumerable there",
);
let sealed = ShellEnforcement::resolve(
&policy,
murmur_artifact::ContainmentClass::Sealed,
HostProbe::probe(),
)
.expect("an empty policy resolves");
let granted: Vec<&LandlockGrant> = sealed
.landlock_grants
.iter()
.filter(|grant| {
crate::sealed::SEALED_RUNTIME_PATHS.contains(&grant.path.to_str().unwrap_or(""))
})
.collect();
if sealed.tier == EnforcementTier::KernelSealed {
assert_eq!(
granted.len(),
crate::sealed::SEALED_RUNTIME_PATHS.len(),
"a sealed session must carry one listable grant per bound runtime path",
);
assert!(granted
.iter()
.all(|grant| grant.list_dir && !grant.executable));
} else {
assert!(
granted.is_empty(),
"this host fell back to {:?}, where there is no composed root to enumerate",
sealed.tier,
);
}
}
#[test]
fn every_sealed_etc_path_gets_a_non_executable_landlock_grant() {
let grants = resolve_sealed_etc_landlock_grants(EnforcementTier::KernelSealed);
assert_eq!(
grants,
crate::sealed::SEALED_ETC_PATHS
.iter()
.map(|entry| LandlockGrant {
path: PathBuf::from(entry.path),
list_dir: entry.directory,
executable: false,
})
.collect::<Vec<_>>(),
"the composed root binds exactly these /etc entries, so exactly these must be \
readable — the bound set and the granted set must not drift apart",
);
assert!(
grants.iter().all(|grant| !grant.executable),
"Execute is the exec allowlist, and nothing in /etc is a program a manifest asked to \
run — /etc/alternatives is a directory of symlinks into /usr/bin",
);
let listable: Vec<&str> = grants
.iter()
.filter(|grant| grant.list_dir)
.map(|grant| grant.path.to_str().unwrap())
.collect();
assert_eq!(
listable,
vec![
"/etc/ld.so.conf.d",
"/etc/alternatives",
"/etc/ssl",
"/etc/pki",
"/etc/ca-certificates",
"/etc/terminfo",
],
"ReadDir belongs to exactly the directory entries: os.listdir('/etc/ssl/certs') and \
the c_rehash symlink lookup need it, and a regular file has no meaningful ReadDir",
);
}
#[test]
fn sealed_etc_landlock_grants_are_sealed_tier_only() {
for tier in [
EnforcementTier::KernelFull,
EnforcementTier::KernelSeccompOnly,
EnforcementTier::EnvironmentOnly,
] {
assert!(
resolve_sealed_etc_landlock_grants(tier).is_empty(),
"{tier:?} has no composed root, so these rules would hand out the real host's \
/etc — its trust store, resolv.conf and account databases",
);
}
}
#[test]
fn shell_enforcement_grants_sealed_etc_paths_only_when_sealed_applies() {
let policy = CapabilityPolicy::default();
let is_etc_grant = |grant: &LandlockGrant| {
crate::sealed::SEALED_ETC_PATHS
.iter()
.any(|entry| Path::new(entry.path) == grant.path)
};
let scoped = ShellEnforcement::resolve(
&policy,
murmur_artifact::ContainmentClass::Scoped,
HostProbe::probe(),
)
.expect("an empty policy resolves");
assert!(
!scoped.landlock_grants.iter().any(is_etc_grant),
"a scoped capsule runs Landlock over the real host filesystem — /etc/shadow's \
neighbours must not become readable there",
);
let sealed = ShellEnforcement::resolve(
&policy,
murmur_artifact::ContainmentClass::Sealed,
HostProbe::probe(),
)
.expect("an empty policy resolves");
let granted: Vec<&LandlockGrant> = sealed
.landlock_grants
.iter()
.filter(|grant| is_etc_grant(grant))
.collect();
if sealed.tier == EnforcementTier::KernelSealed {
assert_eq!(
granted.len(),
crate::sealed::SEALED_ETC_PATHS.len(),
"a sealed session must carry one read grant per bound /etc entry",
);
assert!(granted.iter().all(|grant| !grant.executable));
} else {
assert!(
granted.is_empty(),
"this host fell back to {:?}, where there is no composed root and /etc is the \
host's own",
sealed.tier,
);
}
}
#[test]
fn resolve_network_allowlist_ips_empty_list_is_empty() {
let ips = resolve_network_allowlist_ips(&[]).unwrap();
assert!(ips.is_empty());
}
#[test]
fn resolve_network_allowlist_ips_resolves_localhost_to_loopback() {
let ips = resolve_network_allowlist_ips(&["localhost".to_string()]).unwrap();
assert!(
!ips.is_empty(),
"localhost should resolve to at least one address"
);
assert!(
ips.iter().all(|ip| ip.is_loopback()),
"every resolved localhost address should be loopback, got {ips:?}"
);
}
#[test]
fn resolve_network_allowlist_ips_rejects_invalid_host_syntax_cleanly() {
let error = resolve_network_allowlist_ips(&["".to_string()]).unwrap_err();
assert!(!error.is_empty());
}
#[test]
fn resolve_network_allowlist_ips_skips_unresolvable_host_without_failing() {
let ips =
resolve_network_allowlist_ips(&["definitely-not-a-real-host.invalid".to_string()])
.expect("an unresolvable host must not be a hard error");
assert!(
ips.is_empty(),
"an unresolvable host should contribute no IPs, got {ips:?}"
);
}
#[test]
fn resolve_network_allowlist_ips_keeps_resolvable_hosts_when_another_is_unresolvable() {
let ips = resolve_network_allowlist_ips(&[
"definitely-not-a-real-host.invalid".to_string(),
"localhost".to_string(),
])
.expect("a mix of resolvable and unresolvable hosts must not be a hard error");
assert!(
!ips.is_empty(),
"localhost's addresses should survive alongside an unresolvable host"
);
assert!(
ips.iter().all(|ip| ip.is_loopback()),
"only localhost should have contributed addresses, got {ips:?}"
);
}
#[test]
fn resolve_network_allowlist_ips_dedupes_across_hosts() {
let ips =
resolve_network_allowlist_ips(&["localhost".to_string(), "localhost:8080".to_string()])
.unwrap();
let mut sorted = ips.clone();
sorted.sort();
sorted.dedup();
assert_eq!(
ips.len(),
sorted.len(),
"resolved IPs should already be deduplicated"
);
}
#[test]
fn environment_only_has_no_resolved_ips_and_environment_only_tier() {
let enforcement = ShellEnforcement::environment_only();
assert_eq!(enforcement.tier, EnforcementTier::EnvironmentOnly);
assert!(enforcement.network_allow_ips.is_empty());
}
#[cfg(any(target_os = "linux", target_os = "macos"))]
#[test]
fn nproc_bound_without_a_scope_is_headroom_over_a_live_count() {
match ShellEnforcement::environment_only().nproc_bound() {
crate::resources::NprocBound::Headroom { baseline } => assert!(
baseline > 0,
"the test process is owned by this uid, so the count cannot be zero"
),
other => panic!("no scope must give headroom, got {other:?}"),
}
}
#[cfg(target_os = "linux")]
#[test]
fn nproc_bound_without_a_scope_is_measured_on_each_call() {
let _lock = crate::resources::test_support::host_threads_lock();
let enforcement = ShellEnforcement::environment_only();
let baseline = |bound| match bound {
crate::resources::NprocBound::Headroom { baseline } => baseline,
other => panic!("no scope must give headroom, got {other:?}"),
};
let first = baseline(enforcement.nproc_bound());
let holders = crate::resources::test_support::HeldThreads::start(200);
let second = baseline(enforcement.nproc_bound());
drop(holders);
assert!(
second > first,
"200 threads started between the calls must raise the baseline: {first} then {second}"
);
}
#[cfg(target_os = "linux")]
#[test]
fn nproc_bound_with_a_scope_leaves_rlimit_nproc_to_pids_max() {
if !crate::cgroup::cgroup_delegation_available() {
crate::runtime_err!(
"[SKIP-HOST] nproc_bound_with_a_scope_leaves_rlimit_nproc_to_pids_max: this host \
cannot delegate a cgroup v2 scope"
);
return;
}
let temp = tempfile::tempdir().unwrap();
let limits = crate::resources::HostResourceLimits::default();
let scope = crate::cgroup::prepare_scope(true, &limits, "ses_nproc_bound", temp.path())
.expect("this host delegates a cgroup scope")
.scope;
let enforcement = ShellEnforcement::environment_only().with_host_bounding(scope, None);
assert_eq!(
enforcement.nproc_bound(),
crate::resources::NprocBound::CgroupPidsMax
);
}
#[cfg(target_os = "linux")]
#[test]
fn a_scoped_capsule_keeps_spawning_after_the_host_adds_threads() {
const TEST: &str = "a_scoped_capsule_keeps_spawning_after_the_host_adds_threads";
if !crate::cgroup::cgroup_delegation_available() {
crate::runtime_err!("[SKIP-HOST] {TEST}: this host cannot delegate a cgroup v2 scope");
return;
}
let _lock = crate::resources::test_support::host_threads_lock();
let temp = tempfile::tempdir().unwrap();
let limits = crate::resources::HostResourceLimits {
max_processes: 32,
max_open_files: 256,
..crate::resources::HostResourceLimits::default()
};
let scope = crate::cgroup::prepare_scope(true, &limits, "ses_nproc_race", temp.path())
.expect("this host delegates a cgroup scope")
.scope;
assert!(
scope.is_some(),
"a required scope on Linux is always created"
);
let node_available = crate::sandbox::resolve_invoked_binary_path("node") != "node";
let mut shell_allow = vec!["bash".to_string()];
if node_available {
shell_allow.push("node".to_string());
}
let policy = CapabilityPolicy {
shell_allow,
resources: limits,
..CapabilityPolicy::default()
};
let mut enforcement = ShellEnforcement::environment_only();
enforcement.resource_limits = limits;
let enforcement = enforcement.with_host_bounding(scope, None);
let holders = crate::resources::test_support::HeldThreads::start(400);
let result = crate::shell::execute_shell(
"bash",
&[
"-c",
"for i in $(seq 1 40); do sleep 0.2 & done; wait; ulimit -Hu; ulimit -Hn",
],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.expect("bash must run");
assert_eq!(result.exit_code, 0, "stderr was: {}", result.stderr);
assert_eq!(result.resource_limit_hit, None);
let (_, own_hard) = crate::resources::test_support::own_rlimit_nproc();
let lines: Vec<&str> = result.stdout.lines().collect();
assert_eq!(
lines,
[
crate::resources::test_support::rendered_rlimit(own_hard).as_str(),
"256"
],
"RLIMIT_NPROC is inherited unchanged and RLIMIT_NOFILE is the declared ceiling",
);
if node_available {
let script = "const { spawn } = require('child_process');\n\
const runs = Array.from({ length: 8 }, () => new Promise((resolve) => {\n\
const child = spawn(process.execPath, ['-e', 'process.exit(0)']);\n\
child.once('error', (error) => resolve(error.message));\n\
child.once('exit', (code) => resolve(code === 0 ? null : `exit ${code}`));\n\
}));\n\
Promise.all(runs).then((outcomes) => {\n\
const failures = outcomes.filter((outcome) => outcome !== null);\n\
if (failures.length > 0) { console.error(failures.join('\\n')); process.exit(1); }\n\
});\n";
let result = crate::shell::execute_shell(
"node",
&["-e", script],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.expect("node must run");
assert_eq!(result.exit_code, 0, "stderr was: {}", result.stderr);
} else {
crate::runtime_err!("[SKIP-HOST] {TEST}: no node on PATH, skipping the node half");
}
drop(holders);
}
#[cfg(target_os = "linux")]
#[test]
fn without_a_scope_the_nproc_baseline_is_measured_per_spawn() {
const TEST: &str = "without_a_scope_the_nproc_baseline_is_measured_per_spawn";
let _lock = crate::resources::test_support::host_threads_lock();
let temp = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
resources: crate::resources::HostResourceLimits {
max_processes: 64,
..crate::resources::HostResourceLimits::default()
},
..CapabilityPolicy::default()
};
let mut enforcement = ShellEnforcement::environment_only();
enforcement.resource_limits = policy.resources;
let before = crate::resources::uid_task_count().expect("Linux can count the uid's tasks");
let (_, own_hard) = crate::resources::test_support::own_rlimit_nproc();
if own_hard != libc::RLIM_INFINITY && own_hard < before + 664 {
crate::runtime_err!(
"[SKIP-HOST] {TEST}: the inherited hard RLIMIT_NPROC is {own_hard}, too close to \
the uid's {before} tasks for 300 more threads to show in a child's ceiling"
);
return;
}
let hard_nproc = || -> u64 {
let result = crate::shell::execute_shell(
"bash",
&["-c", "ulimit -Hu"],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.expect("bash must run");
assert_eq!(result.exit_code, 0, "stderr was: {}", result.stderr);
result.stdout.trim().parse().expect("a finite ceiling")
};
let holders = crate::resources::test_support::HeldThreads::start(300);
let held = hard_nproc();
assert!(
held >= before + 64 + 150,
"the ceiling must count the holder threads started after the enforcement was \
resolved: {held} with {before} tasks before them"
);
drop(holders);
let released = hard_nproc();
assert!(
released < held,
"the ceiling must drop once the holders exit: {released} after, {held} while held"
);
}
fn make_executable(path: &Path, contents: &str) {
use std::os::unix::fs::PermissionsExt;
std::fs::write(path, contents).unwrap();
let mut perms = std::fs::metadata(path).unwrap().permissions();
perms.set_mode(0o755);
std::fs::set_permissions(path, perms).unwrap();
}
fn bootstrap_log_contents(workdir: &Path) -> String {
std::fs::read_to_string(workdir.join("logs").join("bootstrap.log")).unwrap_or_default()
}
#[test]
fn resolve_exec_allowlist_resolves_bare_name_via_path_dirs() {
let temp = tempfile::tempdir().unwrap();
let bin_dir = temp.path().join("bin");
std::fs::create_dir(&bin_dir).unwrap();
let tool = bin_dir.join("mytool");
make_executable(&tool, "#!/bin/sh\nexit 0\n");
let resolved = resolve_exec_allowlist_in(&["mytool".to_string()], &[bin_dir]);
assert_eq!(resolved, vec![std::fs::canonicalize(&tool).unwrap()]);
}
#[test]
fn resolve_exec_allowlist_skips_entries_that_do_not_resolve() {
let temp = tempfile::tempdir().unwrap();
let resolved = resolve_exec_allowlist_in(
&["definitely-not-a-real-binary".to_string(), String::new()],
&[temp.path().to_path_buf()],
);
assert!(
resolved.is_empty(),
"unresolvable entries must shrink (not fail) the allow set: {resolved:?}"
);
}
#[test]
fn resolve_exec_allowlist_ignores_non_executable_path_matches() {
let temp = tempfile::tempdir().unwrap();
std::fs::write(temp.path().join("mytool"), "not executable").unwrap();
let resolved =
resolve_exec_allowlist_in(&["mytool".to_string()], &[temp.path().to_path_buf()]);
assert!(resolved.is_empty());
}
#[test]
fn resolve_exec_allowlist_canonicalizes_path_entries_through_symlinks() {
let temp = tempfile::tempdir().unwrap();
let real = temp.path().join("real-tool");
make_executable(&real, "#!/bin/sh\nexit 0\n");
let link = temp.path().join("link-tool");
std::os::unix::fs::symlink(&real, &link).unwrap();
let resolved = resolve_exec_allowlist_in(&[link.to_string_lossy().into_owned()], &[]);
assert_eq!(resolved, vec![std::fs::canonicalize(&real).unwrap()]);
}
#[test]
fn resolve_invoked_binary_path_resolves_bare_name_via_path_dirs() {
let temp = tempfile::tempdir().unwrap();
let bin_dir = temp.path().join("bin");
std::fs::create_dir(&bin_dir).unwrap();
let tool = bin_dir.join("mytool");
make_executable(&tool, "#!/bin/sh\nexit 0\n");
let resolved = resolve_invoked_binary_path_in("mytool", &[bin_dir]);
assert_eq!(
resolved,
std::fs::canonicalize(&tool)
.unwrap()
.to_string_lossy()
.into_owned(),
"a resolvable name must report the canonical absolute path, not the bare name"
);
}
#[test]
fn resolve_invoked_binary_path_falls_back_to_bare_name_when_unresolvable() {
let temp = tempfile::tempdir().unwrap();
let path_dirs = [temp.path().to_path_buf()];
let resolved = resolve_invoked_binary_path_in("definitely-not-a-real-binary", &path_dirs);
assert_eq!(
resolved, "definitely-not-a-real-binary",
"an unresolvable name must fall back to the bare name, never error"
);
assert_eq!(resolve_invoked_binary_path_in("", &[]), "");
}
#[test]
fn resolve_invoked_binary_path_takes_the_first_path_match() {
let temp = tempfile::tempdir().unwrap();
let first = temp.path().join("first");
let second = temp.path().join("second");
std::fs::create_dir(&first).unwrap();
std::fs::create_dir(&second).unwrap();
make_executable(&first.join("mytool"), "#!/bin/sh\nexit 0\n");
make_executable(&second.join("mytool"), "#!/bin/sh\nexit 1\n");
let resolved = resolve_invoked_binary_path_in("mytool", &[first.clone(), second]);
assert_eq!(
resolved,
std::fs::canonicalize(first.join("mytool"))
.unwrap()
.to_string_lossy()
.into_owned(),
"reporting must match execvp: the first PATH hit is the one that ran"
);
}
#[test]
fn resolve_invoked_binary_path_ignores_non_executable_path_matches() {
let temp = tempfile::tempdir().unwrap();
std::fs::write(temp.path().join("mytool"), "not executable").unwrap();
assert_eq!(
resolve_invoked_binary_path_in("mytool", &[temp.path().to_path_buf()]),
"mytool"
);
}
#[test]
fn resolve_invoked_binary_path_canonicalizes_a_path_entry_through_symlinks() {
let temp = tempfile::tempdir().unwrap();
let real = temp.path().join("real-tool");
make_executable(&real, "#!/bin/sh\nexit 0\n");
let link = temp.path().join("link-tool");
std::os::unix::fs::symlink(&real, &link).unwrap();
assert_eq!(
resolve_invoked_binary_path_in(&link.to_string_lossy(), &[]),
std::fs::canonicalize(&real)
.unwrap()
.to_string_lossy()
.into_owned()
);
}
fn build_elf64(interp: Option<&str>, needed: &[&str], runpath: Option<&str>) -> Vec<u8> {
let has_dynamic = !needed.is_empty() || runpath.is_some();
let mut phnum: u16 = 1; if interp.is_some() {
phnum += 1;
}
if has_dynamic {
phnum += 1;
}
let ehsize = 64usize;
let phentsize = 56usize;
let phoff = ehsize;
let body_start = phoff + phentsize * phnum as usize;
let mut strtab: Vec<u8> = vec![0];
let mut needed_offsets = Vec::new();
for name in needed {
needed_offsets.push(strtab.len() as u64);
strtab.extend_from_slice(name.as_bytes());
strtab.push(0);
}
let runpath_offset = runpath.map(|r| {
let off = strtab.len() as u64;
strtab.extend_from_slice(r.as_bytes());
strtab.push(0);
off
});
let mut body: Vec<u8> = Vec::new();
let interp_offset = body_start + body.len();
if let Some(i) = interp {
body.extend_from_slice(i.as_bytes());
body.push(0);
}
while !(body_start + body.len()).is_multiple_of(8) {
body.push(0);
}
let dyn_offset = body_start + body.len();
let num_dyn = needed.len() + runpath.map_or(0, |_| 1) + 3; let dyn_size = num_dyn * 16;
let strtab_offset = dyn_offset + dyn_size;
if has_dynamic {
let mut push_dyn = |tag: u64, val: u64| {
body.extend_from_slice(&tag.to_le_bytes());
body.extend_from_slice(&val.to_le_bytes());
};
for &o in &needed_offsets {
push_dyn(1, o); }
if let Some(ro) = runpath_offset {
push_dyn(29, ro); }
push_dyn(5, strtab_offset as u64); push_dyn(10, strtab.len() as u64); push_dyn(0, 0); assert_eq!(body_start + body.len(), strtab_offset);
body.extend_from_slice(&strtab);
}
let total_len = body_start + body.len();
let mut out = vec![0u8; total_len];
out[0..4].copy_from_slice(b"\x7fELF");
out[4] = 2; out[5] = 1; out[6] = 1; out[16..18].copy_from_slice(&3u16.to_le_bytes()); out[18..20].copy_from_slice(&0x3eu16.to_le_bytes()); out[20..24].copy_from_slice(&1u32.to_le_bytes()); out[32..40].copy_from_slice(&(phoff as u64).to_le_bytes()); out[54..56].copy_from_slice(&(phentsize as u16).to_le_bytes()); out[56..58].copy_from_slice(&phnum.to_le_bytes());
let mut phdrs: Vec<(u32, u64, u64, u64)> = Vec::new(); phdrs.push((1, 0, 0, total_len as u64)); if let Some(i) = interp {
phdrs.push((
3,
interp_offset as u64,
interp_offset as u64,
i.len() as u64 + 1,
));
}
if has_dynamic {
phdrs.push((2, dyn_offset as u64, dyn_offset as u64, dyn_size as u64));
}
for (idx, (p_type, p_offset, p_vaddr, p_filesz)) in phdrs.iter().enumerate() {
let base = phoff + idx * phentsize;
out[base..base + 4].copy_from_slice(&p_type.to_le_bytes());
out[base + 8..base + 16].copy_from_slice(&p_offset.to_le_bytes());
out[base + 16..base + 24].copy_from_slice(&p_vaddr.to_le_bytes());
out[base + 32..base + 40].copy_from_slice(&p_filesz.to_le_bytes());
}
out[body_start..total_len].copy_from_slice(&body);
out
}
#[test]
fn parse_elf_extracts_interp_needed_and_runpath() {
let bytes = build_elf64(
Some("/lib64/ld-linux-x86-64.so.2"),
&["libc.so.6", "libm.so.6"],
Some("$ORIGIN/../lib:/opt/custom/lib"),
);
let deps = parse_elf_dependencies(&bytes).expect("well-formed ELF64 must parse");
assert_eq!(deps.interp.as_deref(), Some("/lib64/ld-linux-x86-64.so.2"));
assert_eq!(
deps.needed,
vec!["libc.so.6".to_string(), "libm.so.6".to_string()]
);
assert_eq!(
deps.runpaths,
vec!["$ORIGIN/../lib".to_string(), "/opt/custom/lib".to_string()],
"DT_RUNPATH must be split on ':' with empty parts dropped"
);
}
#[test]
fn parse_elf_static_binary_contributes_only_itself() {
let bytes = build_elf64(None, &[], None);
let deps = parse_elf_dependencies(&bytes).expect("a static ELF64 is still valid ELF");
assert_eq!(deps, ElfDependencies::default());
}
#[test]
fn parse_elf_rejects_non_elf_and_non_elf64() {
assert!(parse_elf_dependencies(b"not an elf at all, way too short").is_none());
let mut elf32 = build_elf64(Some("/lib/ld.so"), &["libc.so.6"], None);
elf32[4] = 1;
assert!(
parse_elf_dependencies(&elf32).is_none(),
"32-bit ELF is not a binary this runtime execs and must not parse"
);
}
#[test]
fn resolve_soname_finds_via_search_dirs_and_runpath_origin() {
let temp = tempfile::tempdir().unwrap();
let origin = temp.path().join("app");
std::fs::create_dir(&origin).unwrap();
let libs = temp.path().join("libs");
std::fs::create_dir(&libs).unwrap();
std::fs::write(libs.join("libc.so.6"), "x").unwrap();
assert_eq!(
resolve_soname("libc.so.6", &origin, &[], std::slice::from_ref(&libs)),
Some(std::fs::canonicalize(libs.join("libc.so.6")).unwrap())
);
assert_eq!(
resolve_soname("libmissing.so", &origin, &[], std::slice::from_ref(&libs)),
None
);
let origin_lib = origin.join("lib");
std::fs::create_dir(&origin_lib).unwrap();
std::fs::write(origin_lib.join("libfoo.so"), "x").unwrap();
assert_eq!(
resolve_soname("libfoo.so", &origin, &["$ORIGIN/lib".to_string()], &[]),
Some(std::fs::canonicalize(origin_lib.join("libfoo.so")).unwrap())
);
}
#[test]
fn resolve_landlock_grants_derives_interp_and_needed_closure() {
let temp = tempfile::tempdir().unwrap();
let bin_dir = temp.path().join("bin");
let lib_dir = temp.path().join("lib");
std::fs::create_dir(&bin_dir).unwrap();
std::fs::create_dir(&lib_dir).unwrap();
let interp = bin_dir.join("ld-fake.so");
std::fs::write(&interp, "not really elf").unwrap();
std::fs::write(lib_dir.join("libfake.so.1"), "not really elf").unwrap();
let prog = bin_dir.join("prog");
let elf = build_elf64(Some(interp.to_str().unwrap()), &["libfake.so.1"], None);
std::fs::write(&prog, &elf).unwrap();
let grants =
resolve_landlock_grants_in(std::slice::from_ref(&prog), std::slice::from_ref(&lib_dir));
let expect = |p: PathBuf| std::fs::canonicalize(p).unwrap();
assert!(
grants.contains(&expect(prog)),
"the binary itself must be granted: {grants:?}"
);
assert!(
grants.contains(&expect(interp)),
"the ELF interpreter must be granted: {grants:?}"
);
assert!(
grants.contains(&expect(lib_dir.join("libfake.so.1"))),
"each resolved DT_NEEDED library must be granted: {grants:?}"
);
}
#[test]
fn resolve_landlock_grants_skips_unresolvable_deps_and_missing_binaries() {
let temp = tempfile::tempdir().unwrap();
let bin_dir = temp.path().join("bin");
std::fs::create_dir(&bin_dir).unwrap();
let prog = bin_dir.join("prog");
std::fs::write(
&prog,
build_elf64(Some("/definitely/not/here.so"), &["libnope.so"], None),
)
.unwrap();
let missing = bin_dir.join("does-not-exist");
let grants =
resolve_landlock_grants_in(&[prog.clone(), missing], std::slice::from_ref(&bin_dir));
assert_eq!(grants, vec![std::fs::canonicalize(&prog).unwrap()]);
}
fn interp_grant(binary: &str, dirs: &[(&str, bool)]) -> InterpreterRuntimeGrant {
InterpreterRuntimeGrant {
binary: binary.to_string(),
dirs: dirs
.iter()
.map(|(path, list_dir)| murmur_artifact::InterpreterRuntimeDir {
path: path.to_string(),
list_dir: *list_dir,
})
.collect(),
}
}
#[test]
fn resolve_interpreter_runtime_grants_copies_list_dir_verbatim() {
let grants = resolve_interpreter_runtime_grants(&[interp_grant(
"python3",
&[
("/usr/lib/python3.11", true),
("/usr/lib/python3.11/lib-dynload", false),
],
)]);
assert_eq!(
grants,
vec![
LandlockGrant {
path: PathBuf::from("/usr/lib/python3.11"),
list_dir: true,
executable: true,
},
LandlockGrant {
path: PathBuf::from("/usr/lib/python3.11/lib-dynload"),
list_dir: false,
executable: true,
},
]
);
}
#[test]
fn resolve_interpreter_runtime_grants_flattens_multiple_grants_and_needs_no_existing_path() {
let grants = resolve_interpreter_runtime_grants(&[
interp_grant("python3", &[("/opt/py/nonexistent", false)]),
interp_grant("ruby", &[("/opt/rb/also-nonexistent", true)]),
]);
assert_eq!(
grants,
vec![
LandlockGrant {
path: PathBuf::from("/opt/py/nonexistent"),
list_dir: false,
executable: true,
},
LandlockGrant {
path: PathBuf::from("/opt/rb/also-nonexistent"),
list_dir: true,
executable: true,
},
]
);
assert!(resolve_interpreter_runtime_grants(&[]).is_empty());
}
#[test]
fn non_listable_files_marks_every_closure_file_non_enumerable() {
let wrapped = LandlockGrant::non_listable_files(vec![
PathBuf::from("/lib/x86_64-linux-gnu/libc.so.6"),
PathBuf::from("/usr/bin/bash"),
]);
assert!(wrapped.iter().all(|grant| !grant.list_dir), "{wrapped:?}");
assert_eq!(
wrapped.iter().map(|g| g.path.clone()).collect::<Vec<_>>(),
vec![
PathBuf::from("/lib/x86_64-linux-gnu/libc.so.6"),
PathBuf::from("/usr/bin/bash"),
]
);
}
#[test]
fn capsule_device_grants_are_exactly_three_devices_with_only_dev_null_writable() {
let listed: Vec<(&str, bool)> = CAPSULE_DEVICE_GRANTS
.iter()
.map(|device| (device.path, device.writable))
.collect();
assert_eq!(
listed,
vec![
("/dev/null", true),
("/dev/zero", false),
("/dev/urandom", false),
],
"the capsule device set is fixed at exactly these three paths and these access \
levels. /dev/null must stay writable — Python's subprocess.DEVNULL opens it O_RDWR \
and a shell `2>/dev/null` redirect opens it O_WRONLY, so a read-only grant breaks \
both. Widening this list is allowed only on demonstrated workload failure, and the \
failure gets recorded alongside the widening (see the manual acceptance procedure on \
the diagnostics reference page)."
);
}
#[test]
fn capsule_device_grants_exclude_the_devices_deliberately_left_out() {
for excluded in [
"/dev/random",
"/dev/full",
"/dev/tty",
"/dev/console",
"/dev/mem",
"/dev/sda",
"/dev",
] {
assert!(
!CAPSULE_DEVICE_GRANTS
.iter()
.any(|device| device.path == excluded),
"{excluded} must not be in the capsule device set — it is denied by the ordinary \
no-matching-rule path, and adding it here would be the only way to change that"
);
}
}
#[test]
fn capsule_device_grants_name_absolute_paths_under_dev() {
for device in CAPSULE_DEVICE_GRANTS {
assert!(
device.path.starts_with("/dev/"),
"{} must be an absolute path under /dev",
device.path
);
}
}
#[test]
fn denied_socket_domains_denies_unix_by_default() {
let denied = denied_socket_domains(false);
assert!(
denied.contains(&LINUX_AF_UNIX),
"AF_UNIX must be denied when the manifest does not declare \
capabilities.network.unix_sockets — this is the /var/run/docker.sock escape"
);
}
#[test]
fn denied_socket_domains_omits_unix_only_when_explicitly_allowed() {
let denied = denied_socket_domains(true);
assert!(
!denied.contains(&LINUX_AF_UNIX),
"a capsule that declared capabilities.network.unix_sockets: true must get no \
AF_UNIX deny rule at all"
);
}
#[test]
fn denied_socket_domains_always_denies_netlink_and_packet() {
for unix_sockets_allowed in [false, true] {
let denied = denied_socket_domains(unix_sockets_allowed);
assert!(
denied.contains(&LINUX_AF_NETLINK),
"AF_NETLINK must be denied unconditionally (unix_sockets_allowed = \
{unix_sockets_allowed})"
);
assert!(
denied.contains(&LINUX_AF_PACKET),
"AF_PACKET must be denied unconditionally (unix_sockets_allowed = \
{unix_sockets_allowed})"
);
}
}
#[test]
fn denied_socket_domains_never_denies_ip_families() {
for unix_sockets_allowed in [false, true] {
let denied = denied_socket_domains(unix_sockets_allowed);
assert!(!denied.contains(&LINUX_AF_INET_DOMAIN));
assert!(!denied.contains(&LINUX_AF_INET6_DOMAIN));
}
}
#[test]
fn denied_socket_domains_returns_exactly_the_documented_sets() {
assert_eq!(
denied_socket_domains(false),
vec![LINUX_AF_NETLINK, LINUX_AF_PACKET, LINUX_AF_UNIX]
);
assert_eq!(
denied_socket_domains(true),
vec![LINUX_AF_NETLINK, LINUX_AF_PACKET]
);
}
#[cfg(target_os = "linux")]
#[test]
fn linux_socket_domain_constants_match_libc() {
assert_eq!(LINUX_AF_UNIX, libc::AF_UNIX);
assert_eq!(LINUX_AF_NETLINK, libc::AF_NETLINK);
assert_eq!(LINUX_AF_PACKET, libc::AF_PACKET);
assert_eq!(LINUX_AF_INET_DOMAIN, libc::AF_INET);
assert_eq!(LINUX_AF_INET6_DOMAIN, libc::AF_INET6);
}
#[test]
fn socket_domain_allow_and_deny_sets_never_overlap() {
for unix_sockets_allowed in [false, true] {
let denied = denied_socket_domains(unix_sockets_allowed);
for domain in allowed_socket_domains(unix_sockets_allowed) {
assert!(
!denied.contains(&domain),
"socket domain {domain} is both allowed and denied \
(unix_sockets_allowed = {unix_sockets_allowed})"
);
}
}
}
#[test]
fn allowed_socket_domains_takes_unix_back_exactly_when_declared() {
assert_eq!(
allowed_socket_domains(false),
vec![LINUX_AF_INET_DOMAIN, LINUX_AF_INET6_DOMAIN]
);
assert_eq!(
allowed_socket_domains(true),
vec![LINUX_AF_INET_DOMAIN, LINUX_AF_INET6_DOMAIN, LINUX_AF_UNIX]
);
}
#[test]
fn allowlist_contains_no_syscall_that_must_stay_denied() {
for name in SECCOMP_MUST_STAY_DENIED {
assert!(
!SECCOMP_SYSCALL_ALLOWLIST.contains(name),
"{name} is in SECCOMP_MUST_STAY_DENIED but was added to \
SECCOMP_SYSCALL_ALLOWLIST — if this is deliberate, remove it from the deny list \
in the same edit and say why in the commit"
);
}
}
#[test]
fn allowlist_permits_exec_because_landlock_now_decides_it() {
for name in ["execve", "execveat"] {
assert!(
SECCOMP_SYSCALL_ALLOWLIST.contains(&name),
"{name} must reach the kernel — the default action is a deny, so omitting it \
refuses the child's own exec and nothing runs"
);
}
}
#[test]
fn allowlist_excludes_socket_which_is_ruled_on_by_domain() {
assert!(
!SECCOMP_SYSCALL_ALLOWLIST.contains(&"socket"),
"socket must be permitted per-domain by allowed_socket_domains, never unconditionally"
);
}
#[test]
fn allowlist_has_no_duplicates() {
let mut seen = std::collections::HashSet::new();
for name in SECCOMP_SYSCALL_ALLOWLIST {
assert!(seen.insert(*name), "{name} appears twice in the allowlist");
}
}
#[test]
fn allowlist_covers_the_syscalls_no_process_can_run_without() {
for name in [
"read",
"write",
"openat",
"close",
"mmap",
"munmap",
"brk",
"exit_group",
"futex",
"rt_sigreturn",
"clone",
"wait4",
"getpid",
"getdents64",
] {
assert!(
SECCOMP_SYSCALL_ALLOWLIST.contains(&name),
"{name} is missing from the allowlist — no process could run under this filter"
);
}
}
#[test]
fn network_ip_allowed_matches_only_the_resolved_allowlist() {
let allow = vec![IpAddr::from([127, 0, 0, 1])];
assert!(network_ip_allowed(IpAddr::from([127, 0, 0, 1]), &allow));
assert!(!network_ip_allowed(IpAddr::from([8, 8, 8, 8]), &allow));
assert!(
!network_ip_allowed(IpAddr::from([127, 0, 0, 1]), &[]),
"an empty resolved allowlist must deny every destination"
);
}
#[test]
fn tier_warning_kernel_full_warns_when_a_shell_is_declared() {
assert_eq!(
tier_warning(EnforcementTier::KernelFull, false),
Some((W_SEC_005, KERNEL_ENFORCEMENT_NOTICE))
);
assert_eq!(tier_warning(EnforcementTier::KernelFull, true), None);
}
#[test]
fn tier_warning_all_tiers_warn_only_when_shell_allow_is_nonempty() {
assert_eq!(
tier_warning(EnforcementTier::KernelSeccompOnly, false),
Some((W_SEC_002, SECCOMP_ONLY_WARNING))
);
assert_eq!(
tier_warning(EnforcementTier::EnvironmentOnly, false),
Some((W_SEC_001, ENVIRONMENT_ONLY_WARNING))
);
assert_eq!(tier_warning(EnforcementTier::KernelFull, true), None);
assert_eq!(tier_warning(EnforcementTier::KernelSeccompOnly, true), None);
assert_eq!(tier_warning(EnforcementTier::EnvironmentOnly, true), None);
}
#[test]
fn aggregate_bounding_warns_only_off_linux_and_only_with_a_subprocess_route() {
assert_eq!(
aggregate_bounding_warning(false, true, false),
Some((W_SEC_010, NO_AGGREGATE_BOUNDING_WARNING))
);
assert_eq!(
aggregate_bounding_warning(false, false, false),
None,
"a capsule that cannot spawn a subprocess has nothing to bound"
);
assert_eq!(
aggregate_bounding_warning(true, true, false),
None,
"on Linux this case is a refused launch, not a warning"
);
assert_eq!(
aggregate_bounding_warning(true, true, true),
None,
"a scope exists, so there is no gap"
);
}
#[test]
fn aggregate_bounding_warning_states_why_rlimit_nproc_is_insufficient() {
let (_, message) = aggregate_bounding_warning(false, true, false).unwrap();
assert!(message.contains("per-uid"), "message was: {message}");
assert!(message.contains("fork bomb"), "message was: {message}");
assert!(message.contains("pids.max"), "message was: {message}");
}
#[test]
fn warn_for_missing_aggregate_bounding_writes_the_code_and_link_to_bootstrap_log() {
let temp = tempfile::tempdir().unwrap();
warn_for_missing_aggregate_bounding(temp.path(), true, false);
let log = bootstrap_log_contents(temp.path());
if cfg!(target_os = "linux") {
assert!(
log.is_empty(),
"Linux refuses the launch instead of warning: {log}"
);
} else {
assert!(log.contains(W_SEC_010), "log was: {log}");
assert!(
log.contains(&security_warning_link(W_SEC_010)),
"log was: {log}"
);
}
}
fn warning_test_policy() -> CapabilityPolicy {
CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
network_allow: vec!["https://api.example.com".to_string()],
..CapabilityPolicy::default()
}
}
#[test]
fn warn_for_enforcement_tier_kernel_full_logs_scope_when_shell_declared() {
let temp = tempfile::tempdir().unwrap();
warn_for_enforcement_tier(
EnforcementTier::KernelFull,
temp.path(),
&warning_test_policy(),
);
let log = bootstrap_log_contents(temp.path());
assert!(
log.contains(W_SEC_005),
"KernelFull must not be silent — it must state what the tier does and does not cover: {log}"
);
assert!(
log.contains("denied, not absent"),
"must state the scope's documented limit: paths outside the workdir are denied, not \
absent: {log}"
);
assert!(
log.contains("read+execute") && log.contains("outside the workdir"),
"must describe what is now actually granted (narrow, derived read+execute outside \
the workdir), not an unfixed grant bug: {log}"
);
assert!(
log.contains(&security_warning_link(W_SEC_005)),
"must link to the diagnostics doc page: {log}"
);
}
#[test]
fn warn_for_enforcement_tier_kernel_full_silent_without_shell_allow() {
let temp = tempfile::tempdir().unwrap();
warn_for_enforcement_tier(
EnforcementTier::KernelFull,
temp.path(),
&CapabilityPolicy::default(),
);
assert_eq!(
bootstrap_log_contents(temp.path()),
"",
"with no shell declared there is no subprocess to enforce on — stay silent"
);
}
#[test]
fn warn_for_enforcement_tier_seccomp_only_logs_the_landlock_gap() {
let temp = tempfile::tempdir().unwrap();
warn_for_enforcement_tier(
EnforcementTier::KernelSeccompOnly,
temp.path(),
&warning_test_policy(),
);
let log = bootstrap_log_contents(temp.path());
assert!(
log.contains("Landlock"),
"must name the missing primitive: {log}"
);
assert!(
log.contains("filesystem"),
"must name what is not enforced: {log}"
);
assert!(
log.contains(W_SEC_002),
"must carry its warning code: {log}"
);
assert!(
log.contains(&security_warning_link(W_SEC_002)),
"must link to the diagnostics doc page: {log}"
);
}
#[test]
fn warn_for_enforcement_tier_environment_only_logs_permanent_and_bypass_warnings() {
let temp = tempfile::tempdir().unwrap();
warn_for_enforcement_tier(
EnforcementTier::EnvironmentOnly,
temp.path(),
&warning_test_policy(),
);
let log = bootstrap_log_contents(temp.path());
assert!(
log.contains("permanent"),
"must state the tier is permanent, not pending: {log}"
);
assert!(
log.contains("bash"),
"the still-accurate bash/network bypass warning must also fire: {log}"
);
assert!(
log.contains(W_SEC_001),
"must carry the tier warning code: {log}"
);
assert!(
log.contains(W_SEC_003),
"must carry the bash/network bypass warning code: {log}"
);
}
struct ForcePrepareFailureGuard;
impl ForcePrepareFailureGuard {
fn new() -> Self {
FORCE_PREPARE_FAILURE.with(|flag| flag.set(true));
Self
}
}
impl Drop for ForcePrepareFailureGuard {
fn drop(&mut self) {
FORCE_PREPARE_FAILURE.with(|flag| flag.set(false));
}
}
#[test]
fn execute_shell_fails_closed_when_enforcement_setup_fails() {
let temp = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
..CapabilityPolicy::default()
};
let marker = temp.path().join("spawned-anyway.marker");
let script = format!("echo ran > '{}'", marker.display());
{
let _guard = ForcePrepareFailureGuard::new();
let error = crate::shell::execute_shell(
"bash",
&["-c", &script],
&[],
temp.path(),
temp.path(),
&policy,
&ShellEnforcement::environment_only(),
)
.unwrap_err();
assert!(
error.to_string().contains("sandbox"),
"the error must name what failed to initialize: {error}"
);
assert!(
!marker.exists(),
"no subprocess may be spawned at all when enforcement setup fails"
);
}
let result = crate::shell::execute_shell(
"bash",
&["-c", &script],
&[],
temp.path(),
temp.path(),
&policy,
&ShellEnforcement::environment_only(),
)
.unwrap();
assert_eq!(result.exit_code, 0);
assert!(marker.exists());
}
}
#[cfg(target_os = "linux")]
#[cfg(test)]
#[allow(clippy::print_stdout, clippy::print_stderr)]
mod landlock_probe_isolation {
#[test]
fn the_landlock_probe_does_not_restrict_the_calling_process() {
let before = crate::sealed::probe_sealed_support().namespace;
let _ = super::linux_enforce::probe_landlock_full_access();
let after = crate::sealed::probe_sealed_support().namespace;
assert_eq!(
before, after,
"the Landlock probe changed what the sealed probe reports ({before:?} -> {after:?}) — \
it has restricted the calling process instead of a forked child, which permanently \
forbids the mount family to this process and everything it spawns",
);
}
}
#[cfg(all(test, target_os = "linux"))]
#[allow(clippy::print_stdout, clippy::print_stderr)]
mod linux_integration_tests {
use std::path::Path;
use super::*;
fn host_bounding_base() -> ShellEnforcement {
ShellEnforcement::environment_only()
}
fn session_workdir_under(accessible: &Path) -> PathBuf {
let session = accessible.join(".murmur").join("ses_sandbox_tests");
std::fs::create_dir_all(&session).unwrap();
session
}
#[test]
fn workdir_landlock_grant_withholds_device_and_socket_creation() {
use landlock::{Access, AccessFs, ABI};
let all_v1 = AccessFs::from_all(ABI::V1);
let granted = linux_enforce::workdir_access_rights(true);
for withheld in [AccessFs::MakeChar, AccessFs::MakeBlock, AccessFs::MakeSock] {
assert!(
all_v1.contains(withheld),
"{withheld:?} is expected to be part of Landlock ABI v1 — if it is not, this \
test's premise (and the workdir grant's comment) needs revisiting"
);
assert!(
!granted.contains(withheld),
"the workdir grant must withhold {withheld:?}: with the full ABI v1 set declared \
by handle_access, withholding it here is what makes it denied rather than merely \
un-granted"
);
}
assert!(
granted.contains(AccessFs::MakeFifo),
"MakeFifo stays granted — real build tooling creates named pipes in its working tree"
);
let expected = all_v1 & !(AccessFs::MakeChar | AccessFs::MakeBlock | AccessFs::MakeSock);
assert_eq!(
granted, expected,
"the workdir_exec grant must be exactly Landlock ABI v1 minus \
MakeChar/MakeBlock/MakeSock — nothing else may be added or dropped without updating \
this test and the constant's comment together"
);
}
#[test]
fn workdir_execute_right_is_granted_only_when_workdir_exec_is_declared() {
use landlock::AccessFs;
assert!(
!linux_enforce::workdir_access_rights(false).contains(AccessFs::Execute),
"the default must withhold Execute from the workdir — that withholding is the whole \
of what enforces capabilities.shell.allow now that the exec supervisor is gone"
);
assert!(
linux_enforce::workdir_access_rights(true).contains(AccessFs::Execute),
"capabilities.filesystem.workdir_exec: true must actually grant Execute, or the \
compile-and-run workflows it exists for silently stop working"
);
assert_eq!(
linux_enforce::workdir_access_rights(true) & !AccessFs::Execute,
linux_enforce::workdir_access_rights(false),
"workdir_exec must be a one-bit axis — no other right may ride along with it"
);
}
#[test]
fn the_workdir_backed_tmp_store_is_opened_for_landlock_on_the_sealed_tier_only() {
let accessible = tempfile::tempdir().unwrap();
let session = session_workdir_under(accessible.path());
let fds = linux_enforce::open_landlock_fds(
accessible.path(),
&session,
&[],
&[],
false,
EnforcementTier::KernelSealed,
)
.expect("a real, writable workdir must yield sealed Landlock fds");
assert!(
fds.sealed_tmp_fd().is_some(),
"the composed root binds {} at /tmp read-write on every sealed capsule, so its fd must \
always be opened — a missing rule leaves /tmp mounted but EACCES",
crate::sealed::SEALED_TMP_DIR_NAME,
);
assert!(
session.join(crate::sealed::SEALED_TMP_DIR_NAME).is_dir(),
"the /tmp store must be created here: open_landlock_fds runs before build_sealed_root, \
so nothing else has made this directory yet",
);
assert!(
!accessible
.path()
.join(crate::sealed::SEALED_TMP_DIR_NAME)
.exists(),
"enabling sealed must add nothing to the top level of the accessible workdir — that \
directory is the capsule's deliverable",
);
for tier in [
EnforcementTier::KernelFull,
EnforcementTier::KernelSeccompOnly,
EnforcementTier::EnvironmentOnly,
] {
let accessible = tempfile::tempdir().unwrap();
let session = session_workdir_under(accessible.path());
let fds = linux_enforce::open_landlock_fds(
accessible.path(),
&session,
&[],
&[],
false,
tier,
)
.expect("a real, writable workdir must yield Landlock fds on any tier");
assert!(
fds.sealed_tmp_fd().is_none(),
"{tier:?} composes no root and binds nothing at /tmp, so granting a workdir-backed \
/tmp would grant a path this tier's capsule never sees",
);
assert!(
!session.join(crate::sealed::SEALED_TMP_DIR_NAME).exists(),
"{tier:?} must not even create the store — an unbound directory in the session \
workdir is litter the capsule can see",
);
}
}
#[test]
fn the_synthetic_etc_files_are_written_and_opened_on_the_sealed_tier_only() {
let accessible = tempfile::tempdir().unwrap();
let session = session_workdir_under(accessible.path());
let fds = linux_enforce::open_landlock_fds(
accessible.path(),
&session,
&[],
&[],
false,
EnforcementTier::KernelSealed,
)
.expect("a real, writable workdir must yield sealed Landlock fds");
let synthetic: Vec<&crate::sealed::SealedEtcPath> = crate::sealed::SEALED_ETC_PATHS
.iter()
.filter(|entry| entry.synthetic.is_some())
.collect();
assert_eq!(
fds.sealed_identity_fd_count(),
synthetic.len(),
"every synthetic /etc entry the composed root binds needs a rule of its own: {synthetic:?}",
);
for entry in &synthetic {
let path = crate::sealed::synthetic_etc_source(&session, entry.synthetic.unwrap());
assert!(
path.is_file(),
"{} must be written here: open_landlock_fds runs before build_sealed_root, so \
nothing else has written it yet",
entry.path,
);
}
assert!(
!accessible
.path()
.join(crate::sealed::SEALED_ETC_STAGING_DIR_NAME)
.exists(),
"the staging directory belongs to the session, not to the capsule's deliverable",
);
for tier in [
EnforcementTier::KernelFull,
EnforcementTier::KernelSeccompOnly,
EnforcementTier::EnvironmentOnly,
] {
let accessible = tempfile::tempdir().unwrap();
let session = session_workdir_under(accessible.path());
let fds = linux_enforce::open_landlock_fds(
accessible.path(),
&session,
&[],
&[],
false,
tier,
)
.expect("a real, writable workdir must yield Landlock fds on any tier");
assert_eq!(
fds.sealed_identity_fd_count(),
0,
"{tier:?} composes no root, so its capsule reads the host's real /etc/passwd and \
these files stand in for nothing",
);
assert!(
!session
.join(crate::sealed::SEALED_ETC_STAGING_DIR_NAME)
.exists(),
"{tier:?} must not even create the staging directory — same reason as .mur-tmp",
);
}
}
#[test]
fn the_synthetic_passwd_entry_carries_exactly_the_home_the_subprocess_env_gets() {
use std::os::unix::fs::MetadataExt;
let accessible = tempfile::tempdir().unwrap();
let session = session_workdir_under(accessible.path());
let written = ensure_sealed_identity_files(&session).unwrap();
let env = crate::shell::build_shell_env(
&crate::types::CapabilityPolicy::default(),
&[],
&session,
)
.unwrap();
let passwd = std::fs::read_to_string(&written[0]).unwrap();
let owner = std::fs::metadata(&session).unwrap();
let entry: Vec<&str> = passwd
.lines()
.last()
.expect("the capsule's own entry is the last line")
.split(':')
.collect();
assert_eq!(entry[2], owner.uid().to_string(), "uid field");
assert_eq!(entry[3], owner.gid().to_string(), "gid field");
assert_eq!(
Some(&entry[5].to_string()),
env.get("HOME"),
"pw_dir and $HOME are the same string or the entry is worse than useless",
);
let group = std::fs::read_to_string(&written[1]).unwrap();
assert_eq!(
group.lines().last().unwrap().split(':').nth(2),
Some(owner.gid().to_string().as_str()),
"gid field of the capsule's own group",
);
}
#[test]
fn the_composed_root_tmp_rule_carries_exactly_the_workdir_right_set() {
use landlock::AccessFs;
for workdir_exec in [false, true] {
let rights = linux_enforce::workdir_access_rights(workdir_exec);
for needed in [
AccessFs::WriteFile,
AccessFs::ReadFile,
AccessFs::ReadDir,
AccessFs::MakeReg,
AccessFs::MakeDir,
AccessFs::RemoveFile,
AccessFs::RemoveDir,
] {
assert!(
rights.contains(needed),
"/tmp carries the workdir's rights, so it must carry {needed:?}: without it \
`cc`, `mktemp` and `pip` fail on a mount that is genuinely writable",
);
}
assert_eq!(
rights.contains(AccessFs::Execute),
workdir_exec,
"/tmp is backed by a directory inside the workdir, so Execute there must follow \
capabilities.filesystem.workdir_exec exactly — granting it unconditionally would \
make /tmp a hole through which capsule-written bytes run",
);
}
assert_eq!(
linux_enforce::workdir_access_rights(true) & !AccessFs::Execute,
linux_enforce::workdir_access_rights(false),
"the /tmp rule and the workdir rule share one right-set binding in \
apply_landlock_scope; if that set ever grows a second conditional axis, this test and \
that binding must be revisited together",
);
}
#[test]
fn prepare_enforcement_is_noop_for_environment_only_tier_even_on_linux() {
let mut command = std::process::Command::new("true");
let enforcement = ShellEnforcement::environment_only();
let supervisor = prepare_enforcement(
&mut command,
&enforcement,
Path::new("/tmp"),
Path::new("/tmp"),
)
.expect("environment_only must never fail");
supervisor.join_best_effort();
}
#[test]
fn kernel_tier_allows_exec_within_shell_allowlist() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_tier_allows_exec_within_shell_allowlist",
) {
return;
}
let tier = detect_enforcement_tier();
if tier == EnforcementTier::EnvironmentOnly {
eprintln!(
"skipping kernel_tier_allows_exec_within_shell_allowlist: resolved to \
EnvironmentOnly on a Linux host — degrading gracefully"
);
return;
}
let temp = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
..CapabilityPolicy::default()
};
let enforcement = ShellEnforcement {
tier,
network_allow_ips: Vec::new(),
unix_sockets_allowed: policy.unix_sockets_allowed,
landlock_grants: LandlockGrant::non_listable_files(resolve_landlock_grants(
&resolve_exec_allowlist(&policy.shell_allow),
)),
..host_bounding_base()
};
let result = crate::shell::execute_shell(
"bash",
&["-c", "exit 7"],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.unwrap();
assert_eq!(
result.exit_code, 7,
"bash itself is allowlisted, so kernel enforcement must permit its own execve — \
proving the allowlist isn't a blanket deny"
);
}
#[test]
fn kernel_tier_allows_nested_exec_of_second_allowlisted_binary() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_tier_allows_nested_exec_of_second_allowlisted_binary",
) {
return;
}
let tier = detect_enforcement_tier();
if tier == EnforcementTier::EnvironmentOnly {
eprintln!(
"skipping kernel_tier_allows_nested_exec_of_second_allowlisted_binary: \
resolved to EnvironmentOnly on a Linux host — degrading gracefully"
);
return;
}
let temp = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string(), "ls".to_string()],
..CapabilityPolicy::default()
};
let enforcement = ShellEnforcement {
tier,
network_allow_ips: Vec::new(),
unix_sockets_allowed: policy.unix_sockets_allowed,
landlock_grants: LandlockGrant::non_listable_files(resolve_landlock_grants(
&resolve_exec_allowlist(&policy.shell_allow),
)),
..host_bounding_base()
};
let result = crate::shell::execute_shell(
"bash",
&["-c", "ls"],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.unwrap();
assert_eq!(
result.exit_code, 0,
"a second allowlisted binary exec'd from inside allowlisted bash must succeed \
(stderr: {})",
result.stderr
);
}
#[test]
fn kernel_tier_denies_network_connect_outside_allowlist() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_tier_denies_network_connect_outside_allowlist",
) {
return;
}
let tier = detect_enforcement_tier();
if tier == EnforcementTier::EnvironmentOnly {
eprintln!(
"skipping kernel_tier_denies_network_connect_outside_allowlist: resolved to \
EnvironmentOnly on a Linux host — degrading gracefully"
);
return;
}
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let temp = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
..CapabilityPolicy::default()
};
let enforcement = ShellEnforcement {
tier,
network_allow_rules: Vec::new(),
network_allow_ips: Vec::new(),
unix_sockets_allowed: policy.unix_sockets_allowed,
landlock_grants: LandlockGrant::non_listable_files(resolve_landlock_grants(
&resolve_exec_allowlist(&policy.shell_allow),
)),
..host_bounding_base()
};
let script = format!("exec 3<>/dev/tcp/127.0.0.1/{port}");
let result = crate::shell::execute_shell(
"bash",
&["-c", &script],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.unwrap();
assert_ne!(
result.exit_code, 0,
"a direct connect() to a destination outside capabilities.network.allow must fail \
even though the port is really open on the host — the subprocess is in its own \
network namespace with no route to it"
);
}
#[test]
fn kernel_tier_reaches_an_allowlisted_destination_through_the_egress_proxy() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_tier_reaches_an_allowlisted_destination_through_the_egress_proxy",
) {
return;
}
let tier = detect_enforcement_tier();
if tier == EnforcementTier::EnvironmentOnly {
eprintln!(
"skipping kernel_tier_reaches_an_allowlisted_destination_through_the_egress_proxy: \
resolved to EnvironmentOnly on a Linux host — degrading gracefully"
);
return;
}
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let server = std::thread::spawn(move || {
if let Ok((mut stream, _)) = listener.accept() {
use std::io::Write;
let _ = stream.write_all(b"pong\n");
}
});
let temp = tempfile::tempdir().unwrap();
let allow = vec![format!("http://127.0.0.1:{port}")];
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
network_allow: allow.clone(),
..CapabilityPolicy::default()
};
let rules = crate::network_policy::parse_network_allow_rules(&allow).unwrap();
let enforcement = ShellEnforcement {
tier,
egress_tcp_ports: crate::egress_proxy::egress_listen_ports(&rules),
network_allow_rules: rules,
network_allow_ips: resolve_network_allowlist_ips(&allow).unwrap(),
unix_sockets_allowed: policy.unix_sockets_allowed,
landlock_grants: LandlockGrant::non_listable_files(resolve_landlock_grants(
&resolve_exec_allowlist(&policy.shell_allow),
)),
..host_bounding_base()
};
let script =
format!("exec 3<>/dev/tcp/127.0.0.1/{port} && read -r reply <&3 && echo \"$reply\"");
let result = crate::shell::execute_shell(
"bash",
&["-c", &script],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.unwrap();
let _ = server.join();
assert_eq!(
result.exit_code, 0,
"a destination inside capabilities.network.allow must stay reachable — enforcement \
adds denials for out-of-policy destinations, it does not degrade permitted ones \
(stderr: {})",
result.stderr
);
assert!(
result.stdout.contains("pong"),
"the host listener's bytes must come back through the proxy unmodified \
(stdout: {:?}, stderr: {})",
result.stdout,
result.stderr
);
}
#[test]
fn kernel_full_denies_filesystem_access_outside_workdir() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_full_denies_filesystem_access_outside_workdir",
) {
return;
}
let tier = detect_enforcement_tier();
if tier != EnforcementTier::KernelFull {
eprintln!(
"skipping kernel_full_denies_filesystem_access_outside_workdir: requires \
EnforcementTier::KernelFull (Landlock ABI, kernel 5.13+); detected {tier:?} \
instead"
);
return;
}
let temp = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
..CapabilityPolicy::default()
};
let enforcement = ShellEnforcement {
tier,
network_allow_ips: Vec::new(),
unix_sockets_allowed: policy.unix_sockets_allowed,
landlock_grants: LandlockGrant::non_listable_files(resolve_landlock_grants(
&resolve_exec_allowlist(&policy.shell_allow),
)),
..host_bounding_base()
};
let result = crate::shell::execute_shell(
"bash",
&["-c", "cat /etc/hostname"],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.unwrap();
assert_ne!(
result.exit_code, 0,
"reading a file outside the Landlock-scoped workdir must fail under KernelFull"
);
}
#[test]
fn kernel_full_runs_nontrivial_shell_allowlist_but_a_pass_here_does_not_prove_the_landlock_fix()
{
if crate::network_namespace::skip_without_egress_namespace("kernel_full_runs_nontrivial_shell_allowlist_but_a_pass_here_does_not_prove_the_landlock_fix") {
return;
}
let tier = detect_enforcement_tier();
if tier != EnforcementTier::KernelFull {
eprintln!(
"SKIP — PROVES NOTHING ABOUT THE LANDLOCK FIX ON THIS HOST: \
kernel_full_runs_nontrivial_shell_allowlist_... requires \
EnforcementTier::KernelFull (Landlock ABI, kernel 5.13+); detected {tier:?}. \
This run does NOT execute the derived-grant Landlock code path at all, so a \
green result here is not evidence the workdir-scope exec bug is fixed — only a \
real KernelFull Linux host can prove that."
);
return;
}
let temp = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string(), "cat".to_string()],
..CapabilityPolicy::default()
};
let exec_allow_paths = resolve_exec_allowlist(&policy.shell_allow);
let enforcement = ShellEnforcement {
tier,
network_allow_ips: Vec::new(),
unix_sockets_allowed: policy.unix_sockets_allowed,
landlock_grants: LandlockGrant::non_listable_files(resolve_landlock_grants(
&exec_allow_paths,
)),
..host_bounding_base()
};
let result = crate::shell::execute_shell(
"bash",
&["-c", "echo landlock-grant-ok | cat"],
&[],
temp.path(),
temp.path(),
&policy,
&enforcement,
)
.expect("execute_shell should return Ok, not Err");
assert_eq!(
result.exit_code, 0,
"bash + cat are both allowlisted and dynamically linked; the derived Landlock grants \
must let each exec and dynamic-link outside the workdir (stderr: {})",
result.stderr
);
assert!(
result.stdout.contains("landlock-grant-ok"),
"the nested `cat` must actually run and pass its stdin through (stdout: {}, stderr: {})",
result.stdout,
result.stderr
);
}
#[test]
fn kernel_full_denies_write_outside_workdir_but_a_pass_here_does_not_prove_the_landlock_fix() {
if crate::network_namespace::skip_without_egress_namespace("kernel_full_denies_write_outside_workdir_but_a_pass_here_does_not_prove_the_landlock_fix") {
return;
}
let tier = detect_enforcement_tier();
if tier != EnforcementTier::KernelFull {
eprintln!(
"SKIP — PROVES NOTHING ABOUT THE LANDLOCK FIX ON THIS HOST: \
kernel_full_denies_write_outside_workdir_... requires \
EnforcementTier::KernelFull (Landlock ABI, kernel 5.13+); detected {tier:?}. \
This run does NOT execute the Landlock code path, so a green result here is not \
evidence writes outside the workdir are actually denied — only a real KernelFull \
Linux host can prove that."
);
return;
}
let workdir = tempfile::tempdir().unwrap();
let outside = tempfile::tempdir().unwrap();
let target = outside.path().join("should-not-be-written.txt");
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
..CapabilityPolicy::default()
};
let exec_allow_paths = resolve_exec_allowlist(&policy.shell_allow);
let enforcement = ShellEnforcement {
tier,
network_allow_ips: Vec::new(),
unix_sockets_allowed: policy.unix_sockets_allowed,
landlock_grants: LandlockGrant::non_listable_files(resolve_landlock_grants(
&exec_allow_paths,
)),
..host_bounding_base()
};
let script = format!("echo pwned > '{}'", target.display());
let result = crate::shell::execute_shell(
"bash",
&["-c", &script],
&[],
workdir.path(),
workdir.path(),
&policy,
&enforcement,
)
.expect("execute_shell should return Ok with a nonzero exit code, not Err");
assert!(
!target.exists(),
"a write to a path outside the Landlock-scoped workdir must be denied under KernelFull"
);
assert_ne!(
result.exit_code, 0,
"bash's redirection to an outside path must fail (Landlock denies the open-for-write)"
);
}
fn kernel_full_enforcement(
tier: EnforcementTier,
policy: &CapabilityPolicy,
) -> ShellEnforcement {
let exec_allow_paths = resolve_exec_allowlist(&policy.shell_allow);
let mut landlock_grants =
LandlockGrant::non_listable_files(resolve_landlock_grants(&exec_allow_paths));
landlock_grants.extend(resolve_interpreter_runtime_grants(
&policy.shell_interpreter_runtime,
));
ShellEnforcement {
tier,
network_allow_ips: Vec::new(),
unix_sockets_allowed: policy.unix_sockets_allowed,
landlock_grants,
..host_bounding_base()
}
}
#[test]
fn a_sealed_command_finds_tmp_and_etc_after_they_moved_into_the_session_workdir() {
if crate::network_namespace::skip_without_egress_namespace(
"a_sealed_command_finds_tmp_and_etc_after_they_moved_into_the_session_workdir",
) {
return;
}
let probe = HostProbe::probe();
if probe.tier() != EnforcementTier::KernelSealed {
eprintln!(
"SKIP — PROVES NOTHING ABOUT THE SEALED ROOT ON THIS HOST: \
a_sealed_command_finds_tmp_and_etc_... needs a host that reaches KernelSealed \
(unprivileged user namespaces plus a usable Landlock ABI); detected {:?}. No \
composed root is built in this run.",
probe.tier()
);
return;
}
let accessible = tempfile::tempdir().unwrap();
let session = accessible.path().join(".murmur").join("ses_sealed_e2e");
std::fs::create_dir_all(&session).unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string(), "cat".to_string(), "id".to_string()],
..CapabilityPolicy::default()
};
let enforcement =
ShellEnforcement::resolve(&policy, murmur_artifact::ContainmentClass::Sealed, probe)
.expect("a sealed-capable host resolves sealed enforcement");
assert_eq!(enforcement.tier, EnforcementTier::KernelSealed);
let result = crate::shell::execute_shell(
"bash",
&["-c", "echo t > /tmp/probe && cat /tmp/probe && id -un"],
&[],
accessible.path(),
&session,
&policy,
&enforcement,
)
.expect("execute_shell returns Ok even when the command itself fails");
assert_eq!(
result.exit_code, 0,
"stdout: {} stderr: {}",
result.stdout, result.stderr
);
let mut lines = result.stdout.lines();
assert_eq!(
lines.next(),
Some("t"),
"the capsule's /tmp must be writable"
);
assert!(
lines.next().is_some_and(|name| !name.is_empty()),
"`id -un` resolves only through the synthetic /etc/passwd the composed root binds: \
{}",
result.stdout
);
assert_eq!(
std::fs::read_to_string(
session
.join(crate::sealed::SEALED_TMP_DIR_NAME)
.join("probe")
)
.unwrap(),
"t\n",
"the bytes written to /tmp land in the session workdir's own store"
);
assert!(session
.join(crate::sealed::SEALED_ETC_STAGING_DIR_NAME)
.join("passwd")
.is_file());
for name in [
crate::sealed::SEALED_TMP_DIR_NAME,
crate::sealed::SEALED_ETC_STAGING_DIR_NAME,
crate::shell::SYNTHETIC_HOME_DIR_NAME,
] {
assert!(
!accessible.path().join(name).exists(),
"enabling sealed must add nothing to the top level of the accessible workdir, \
and it added {name}"
);
}
}
#[test]
fn kernel_full_interpreter_runtime_list_dir_false_opens_file_but_denies_listing() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_full_interpreter_runtime_list_dir_false_opens_file_but_denies_listing",
) {
return;
}
let tier = detect_enforcement_tier();
if tier != EnforcementTier::KernelFull {
eprintln!(
"SKIP — PROVES NOTHING ABOUT THE LANDLOCK FIX ON THIS HOST: \
kernel_full_interpreter_runtime_list_dir_false_... requires \
EnforcementTier::KernelFull (Landlock ABI, kernel 5.13+); detected {tier:?}. \
This run does NOT execute the interpreter_runtime Landlock code path, so a green \
result here is not evidence list_dir:false behaves correctly — only a real \
KernelFull Linux host can prove that."
);
return;
}
let workdir = tempfile::tempdir().unwrap();
let stdlib = tempfile::tempdir().unwrap();
let module = stdlib.path().join("mod.py");
std::fs::write(&module, "x = 1\n").unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string(), "cat".to_string(), "ls".to_string()],
shell_interpreter_runtime: vec![InterpreterRuntimeGrant {
binary: "bash".to_string(),
dirs: vec![murmur_artifact::InterpreterRuntimeDir {
path: stdlib.path().to_string_lossy().into_owned(),
list_dir: false,
}],
}],
..CapabilityPolicy::default()
};
let enforcement = kernel_full_enforcement(tier, &policy);
let read = crate::shell::execute_shell(
"bash",
&["-c", &format!("cat '{}'", module.display())],
&[],
workdir.path(),
workdir.path(),
&policy,
&enforcement,
)
.expect("execute_shell should return Ok");
assert_eq!(
read.exit_code, 0,
"list_dir:false must still permit opening a known file inside the grant (stderr: {})",
read.stderr
);
assert!(read.stdout.contains("x = 1"), "stdout: {}", read.stdout);
let list = crate::shell::execute_shell(
"bash",
&["-c", &format!("ls '{}'", stdlib.path().display())],
&[],
workdir.path(),
workdir.path(),
&policy,
&enforcement,
)
.expect("execute_shell should return Ok with a nonzero exit code");
assert_ne!(
list.exit_code, 0,
"list_dir:false must deny enumerating the granted directory's own entries"
);
}
#[test]
fn kernel_full_interpreter_runtime_list_dir_true_lists_dir_but_not_its_parent() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_full_interpreter_runtime_list_dir_true_lists_dir_but_not_its_parent",
) {
return;
}
let tier = detect_enforcement_tier();
if tier != EnforcementTier::KernelFull {
eprintln!(
"SKIP — PROVES NOTHING ABOUT THE LANDLOCK FIX ON THIS HOST: \
kernel_full_interpreter_runtime_list_dir_true_... requires \
EnforcementTier::KernelFull (Landlock ABI, kernel 5.13+); detected {tier:?}. \
This run does NOT execute the interpreter_runtime Landlock code path, so a green \
result here is not evidence list_dir:true (and its non-widening to the parent) \
behaves correctly — only a real KernelFull Linux host can prove that."
);
return;
}
let workdir = tempfile::tempdir().unwrap();
let parent = tempfile::tempdir().unwrap();
let stdlib = parent.path().join("stdlib");
std::fs::create_dir(&stdlib).unwrap();
std::fs::write(stdlib.join("mod.py"), "x = 1\n").unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string(), "ls".to_string()],
shell_interpreter_runtime: vec![InterpreterRuntimeGrant {
binary: "bash".to_string(),
dirs: vec![murmur_artifact::InterpreterRuntimeDir {
path: stdlib.to_string_lossy().into_owned(),
list_dir: true,
}],
}],
..CapabilityPolicy::default()
};
let enforcement = kernel_full_enforcement(tier, &policy);
let list = crate::shell::execute_shell(
"bash",
&["-c", &format!("ls '{}'", stdlib.display())],
&[],
workdir.path(),
workdir.path(),
&policy,
&enforcement,
)
.expect("execute_shell should return Ok");
assert_eq!(
list.exit_code, 0,
"list_dir:true must permit enumerating the granted directory (stderr: {})",
list.stderr
);
assert!(list.stdout.contains("mod.py"), "stdout: {}", list.stdout);
let list_parent = crate::shell::execute_shell(
"bash",
&["-c", &format!("ls '{}'", parent.path().display())],
&[],
workdir.path(),
workdir.path(),
&policy,
&enforcement,
)
.expect("execute_shell should return Ok with a nonzero exit code");
assert_ne!(
list_parent.exit_code, 0,
"a grant on stdlib/ must not make its (ungranted) parent enumerable"
);
}
#[test]
fn kernel_full_non_executable_grant_lists_its_tree_but_refuses_to_run_from_it() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_full_non_executable_grant_lists_its_tree_but_refuses_to_run_from_it",
) {
return;
}
let host_tier = detect_enforcement_tier();
if !matches!(
host_tier,
EnforcementTier::KernelFull | EnforcementTier::KernelSealed
) {
eprintln!(
"SKIP — PROVES NOTHING ABOUT THE LANDLOCK FIX ON THIS HOST: \
kernel_full_non_executable_grant_... needs a usable Landlock ABI (kernel 5.13+); \
detected {host_tier:?}. This run does NOT install a Landlock domain, so a green \
result is not evidence that withholding Execute denies execve."
);
return;
}
let workdir = tempfile::tempdir().unwrap();
let tree = tempfile::tempdir().unwrap();
std::fs::write(tree.path().join("entry.txt"), "x\n").unwrap();
let runme = tree.path().join("runme");
std::fs::copy("/bin/echo", &runme).unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string(), "ls".to_string()],
..CapabilityPolicy::default()
};
let with_tree_grant = |executable: bool| {
let mut enforcement = kernel_full_enforcement(EnforcementTier::KernelFull, &policy);
enforcement.landlock_grants.push(LandlockGrant {
path: tree.path().to_path_buf(),
list_dir: true,
executable,
});
enforcement
};
let run = |enforcement: &ShellEnforcement, command: String| {
crate::shell::execute_shell(
"bash",
&["-c", &command],
&[],
workdir.path(),
workdir.path(),
&policy,
enforcement,
)
.expect("execute_shell should return Ok even when the command itself fails")
};
let denied = with_tree_grant(false);
let list = run(&denied, format!("ls '{}'", tree.path().display()));
assert_eq!(
list.exit_code, 0,
"a non-executable grant must still be enumerable — that is its entire purpose \
(stderr: {})",
list.stderr
);
assert!(list.stdout.contains("entry.txt"), "stdout: {}", list.stdout);
let exec_denied = run(&denied, format!("'{}' ran", runme.display()));
assert_ne!(
exec_denied.exit_code, 0,
"withholding Execute must deny execve inside the granted tree; a program ran instead \
(stdout: {})",
exec_denied.stdout
);
assert!(
!exec_denied.stdout.contains("ran"),
"the binary produced output, so it executed: {}",
exec_denied.stdout
);
let allowed = with_tree_grant(true);
let exec_allowed = run(&allowed, format!("'{}' ran", runme.display()));
assert_eq!(
exec_allowed.exit_code, 0,
"the same tree granted Execute must run (stderr: {})",
exec_allowed.stderr
);
assert!(
exec_allowed.stdout.contains("ran"),
"stdout: {}",
exec_allowed.stdout
);
}
#[test]
fn kernel_full_non_executable_file_grant_is_readable_and_survives_a_bad_list_dir_claim() {
if crate::network_namespace::skip_without_egress_namespace(
"kernel_full_non_executable_file_grant_is_readable_and_survives_a_bad_list_dir_claim",
) {
return;
}
let host_tier = detect_enforcement_tier();
if !matches!(
host_tier,
EnforcementTier::KernelFull | EnforcementTier::KernelSealed
) {
eprintln!(
"SKIP — PROVES NOTHING ABOUT THE LANDLOCK FIX ON THIS HOST: \
kernel_full_non_executable_file_grant_... needs a usable Landlock ABI (kernel \
5.13+); detected {host_tier:?}. This run does NOT install a Landlock domain, so a \
green result is not evidence that a file grant reads, nor that a ReadDir claim on \
a file is narrowed instead of refusing the launch."
);
return;
}
let workdir = tempfile::tempdir().unwrap();
let store = tempfile::tempdir().unwrap();
let bundle = store.path().join("ca-certificates.crt");
std::fs::write(&bundle, "-----BEGIN CERTIFICATE-----\n").unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string(), "cat".to_string()],
..CapabilityPolicy::default()
};
let with_file_grant = |list_dir: bool| {
let mut enforcement = kernel_full_enforcement(EnforcementTier::KernelFull, &policy);
enforcement.landlock_grants.push(LandlockGrant {
path: bundle.clone(),
list_dir,
executable: false,
});
enforcement
};
let read = |enforcement: &ShellEnforcement| {
crate::shell::execute_shell(
"bash",
&["-c", &format!("cat '{}'", bundle.display())],
&[],
workdir.path(),
workdir.path(),
&policy,
enforcement,
)
.expect("execute_shell should return Ok even when the command itself fails")
};
let honest = read(&with_file_grant(false));
assert_eq!(
honest.exit_code, 0,
"ReadFile alone must open a granted regular file — this is what the /etc allowlist's \
file entries rely on (stderr: {})",
honest.stderr
);
assert!(
honest.stdout.contains("BEGIN CERTIFICATE"),
"stdout: {}",
honest.stdout
);
let overclaimed = read(&with_file_grant(true));
assert_eq!(
overclaimed.exit_code, 0,
"a list_dir:true claim on a regular file must be narrowed to ReadFile, not turned \
into an EINVAL that refuses the launch (stderr: {})",
overclaimed.stderr
);
assert!(
overclaimed.stdout.contains("BEGIN CERTIFICATE"),
"stdout: {}",
overclaimed.stdout
);
}
struct ForceNoNewPrivsFailureGuard;
impl ForceNoNewPrivsFailureGuard {
fn new() -> Self {
FORCE_NO_NEW_PRIVS_FAILURE.with(|flag| flag.set(true));
Self
}
}
impl Drop for ForceNoNewPrivsFailureGuard {
fn drop(&mut self) {
FORCE_NO_NEW_PRIVS_FAILURE.with(|flag| flag.set(false));
}
}
struct ForceLandlockFailureGuard;
impl ForceLandlockFailureGuard {
fn new() -> Self {
FORCE_LANDLOCK_FAILURE.with(|flag| flag.set(true));
Self
}
}
impl Drop for ForceLandlockFailureGuard {
fn drop(&mut self) {
FORCE_LANDLOCK_FAILURE.with(|flag| flag.set(false));
}
}
fn kernel_full_empty_grants() -> ShellEnforcement {
ShellEnforcement {
tier: EnforcementTier::KernelFull,
network_allow_ips: Vec::new(),
unix_sockets_allowed: false,
landlock_grants: Vec::new(),
..host_bounding_base()
}
}
fn workdir_resolution_error() -> String {
let temp = tempfile::tempdir().unwrap();
let missing = temp.path().join("does-not-exist-workdir");
let marker = temp.path().join("never-spawned.marker");
let mut command = std::process::Command::new("bash");
command
.arg("-c")
.arg(format!("echo ran > '{}'", marker.display()));
let error = prepare_enforcement(
&mut command,
&kernel_full_empty_grants(),
&missing,
&missing,
)
.expect_err("an unresolvable workdir must fail before fork()");
assert!(
!marker.exists(),
"prepare_enforcement must never spawn a subprocess: {error}"
);
error
}
fn child_setup_failure_error<G>(arm: impl FnOnce() -> G) -> String {
child_setup_failure_typed(&kernel_full_empty_grants(), arm).to_string()
}
fn child_setup_failure_typed<G>(
enforcement: &ShellEnforcement,
arm: impl FnOnce() -> G,
) -> crate::shell::ShellExecError {
let temp = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
..CapabilityPolicy::default()
};
let marker = temp.path().join("spawned-anyway.marker");
let script = format!("echo ran > '{}'", marker.display());
let _guard = arm();
let error = crate::shell::execute_shell(
"bash",
&["-c", &script],
&[],
temp.path(),
temp.path(),
&policy,
enforcement,
)
.expect_err("a forced child-side setup failure must make execute_shell return Err");
assert!(
!marker.exists(),
"no subprocess may run when pre_exec setup fails: {error}"
);
error
}
#[test]
fn pre_exec_workdir_resolution_failure_is_caught_before_fork() {
let error = workdir_resolution_error();
assert!(
error.contains("workdir") && error.contains("does-not-exist-workdir"),
"the error must name the workdir path that failed to resolve: {error}"
);
assert!(
!error.contains("os error 22"),
"must not collapse to the bare EINVAL string: {error}"
);
}
#[test]
fn pre_exec_no_new_privs_failure_is_distinct_and_fails_closed() {
if crate::network_namespace::skip_without_egress_namespace(
"pre_exec_no_new_privs_failure_is_distinct_and_fails_closed",
) {
return;
}
let error = child_setup_failure_error(ForceNoNewPrivsFailureGuard::new);
assert!(
error.contains("no_new_privs"),
"the error must name the no_new_privs step specifically: {error}"
);
assert!(
!error.contains("os error 22"),
"must not collapse to the bare EINVAL string: {error}"
);
}
#[test]
fn pre_exec_landlock_failure_is_distinct_and_fails_closed() {
if crate::network_namespace::skip_without_egress_namespace(
"pre_exec_landlock_failure_is_distinct_and_fails_closed",
) {
return;
}
let error = child_setup_failure_error(ForceLandlockFailureGuard::new);
assert!(
error.contains("landlock"),
"the error must name the Landlock construction step specifically: {error}"
);
assert!(
!error.contains("os error 22"),
"must not collapse to the bare EINVAL string: {error}"
);
}
#[test]
fn pre_exec_setup_failures_produce_pairwise_distinct_messages() {
if crate::network_namespace::skip_without_egress_namespace(
"pre_exec_setup_failures_produce_pairwise_distinct_messages",
) {
return;
}
let workdir_msg = workdir_resolution_error();
let no_new_privs_msg = child_setup_failure_error(ForceNoNewPrivsFailureGuard::new);
let landlock_msg = child_setup_failure_error(ForceLandlockFailureGuard::new);
let messages = [
("workdir", &workdir_msg),
("no_new_privs", &no_new_privs_msg),
("landlock", &landlock_msg),
];
for (i, (left_name, left)) in messages.iter().enumerate() {
for (right_name, right) in &messages[i + 1..] {
assert_ne!(left, right, "{left_name} vs {right_name} must differ");
}
}
for message in [&workdir_msg, &no_new_privs_msg, &landlock_msg] {
assert!(!message.is_empty(), "a distinct message must not be empty");
assert!(
!message.contains("os error 22"),
"no failure may read as a bare EINVAL: {message}"
);
}
}
#[test]
fn a_composed_root_failure_is_typed_and_session_fatal_not_just_another_message() {
if crate::network_namespace::skip_without_egress_namespace(
"a_composed_root_failure_is_typed_and_session_fatal_not_just_another_message",
) {
return;
}
let error =
child_setup_failure_typed(&sealed_test_enforcement(), ForceSealedRootFailureGuard::new);
assert!(
matches!(
error,
crate::shell::ShellExecError::SealedRootConstructionFailed { .. }
),
"a sealed-root diagnostic must survive as its own variant, not collapse into \
ShellExecError::Failed: {error}"
);
let fatal = error
.session_fatal()
.expect("a composed-root failure must end the session, not just the tool call");
assert!(
matches!(
fatal,
crate::errors::RuntimeError::SealedRootConstructionFailed { .. }
),
"must map to the RuntimeError the CLI renders as E-RUN-014: {fatal}"
);
let rendered = error.to_string();
assert!(
rendered.contains("composed root") && rendered.contains("unshare"),
"the message must name both the mechanism and the step that failed: {rendered}"
);
assert!(
!rendered.contains("os error 22"),
"must not collapse to the bare EINVAL string: {rendered}"
);
}
#[test]
fn an_ordinary_pre_exec_failure_is_not_session_fatal() {
let landlock =
child_setup_failure_typed(&kernel_full_empty_grants(), ForceLandlockFailureGuard::new);
assert!(
landlock.session_fatal().is_none(),
"a Landlock setup failure is the tool call's problem, not the session's: {landlock}"
);
let no_new_privs = child_setup_failure_typed(
&kernel_full_empty_grants(),
ForceNoNewPrivsFailureGuard::new,
);
assert!(
no_new_privs.session_fatal().is_none(),
"a no_new_privs failure is the tool call's problem, not the session's: {no_new_privs}"
);
}
}