use murmur_artifact::{ContainmentClass, ExportMode, Exports, FileExport, PeerFilesExport};
use serde::Serialize;
use crate::{
cgroup::{CgroupScopeReport, CgroupScopeStatus, IoMaxReport, IoMaxStatus},
errors::RuntimeError,
runtime_dir::RuntimeDirPlacement,
sandbox::{EnforcementTier, HostProbe},
sealed::{SealedShortfall, UsernsGrant},
types::CapabilityPolicy,
};
pub use crate::sandbox::{detect_sealed_shortfall, detect_userns_grant};
pub(crate) fn achieved_class_for_tier(tier: EnforcementTier) -> ContainmentClass {
match tier {
EnforcementTier::KernelSealed => ContainmentClass::Sealed,
EnforcementTier::KernelFull => ContainmentClass::Scoped,
EnforcementTier::KernelSeccompOnly | EnforcementTier::EnvironmentOnly => {
ContainmentClass::Advisory
}
EnforcementTier::Unconfined => ContainmentClass::Unconfined,
}
}
fn achieved_containment_class(
tier: EnforcementTier,
workdir_exec: bool,
runtime_dir_inside_workdir: bool,
) -> ContainmentClass {
let class = achieved_class_for_tier(tier);
if workdir_exec || runtime_dir_inside_workdir {
class.min(ContainmentClass::Advisory)
} else {
class
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ContainmentDecision {
declared: ContainmentClass,
host_tier: EnforcementTier,
applied_tier: EnforcementTier,
achieved: ContainmentClass,
}
impl ContainmentDecision {
pub(crate) fn decide(
host_tier: EnforcementTier,
declared: ContainmentClass,
workdir_exec: bool,
runtime_dir_inside_workdir: bool,
) -> Self {
let applied_tier = crate::sandbox::applied_tier(host_tier, declared);
Self {
declared,
host_tier,
applied_tier,
achieved: achieved_containment_class(
applied_tier,
workdir_exec,
runtime_dir_inside_workdir,
),
}
}
pub fn declared(&self) -> ContainmentClass {
self.declared
}
pub fn achieved(&self) -> ContainmentClass {
self.achieved
}
pub fn host_containment(&self) -> ContainmentClass {
achieved_class_for_tier(self.host_tier)
}
pub fn enforcement_tier(&self) -> &'static str {
enforcement_tier_name(self.applied_tier)
}
pub(crate) fn applied_tier(&self) -> EnforcementTier {
self.applied_tier
}
}
pub fn detect_containment_decision(
declared: ContainmentClass,
) -> (ContainmentDecision, Option<SealedShortfall>) {
let probe = HostProbe::probe();
(
ContainmentDecision::decide(probe.tier(), declared, false, false),
probe.sealed_shortfall(),
)
}
pub fn containment_shortfall_reason(
declared: ContainmentClass,
achieved: ContainmentClass,
sealed_shortfall: Option<&SealedShortfall>,
workdir_exec: bool,
runtime_dir_inside: Option<&RuntimeDirPlacement>,
) -> Option<String> {
containment_shortfall(
declared,
achieved,
sealed_shortfall,
workdir_exec,
runtime_dir_inside,
)
.map(|(reason, _)| reason)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ShortfallCause {
Host,
WorkdirExec,
InsideWorkdir,
}
fn containment_shortfall(
declared: ContainmentClass,
achieved: ContainmentClass,
sealed_shortfall: Option<&SealedShortfall>,
workdir_exec: bool,
runtime_dir_inside: Option<&RuntimeDirPlacement>,
) -> Option<(String, ShortfallCause)> {
if achieved >= declared {
return None;
}
if workdir_exec && declared > ContainmentClass::Advisory {
return Some((
WORKDIR_EXEC_SHORTFALL.to_string(),
ShortfallCause::WorkdirExec,
));
}
if let Some(placement) = runtime_dir_inside.filter(|_| declared > ContainmentClass::Advisory) {
return Some((placement.inside_shortfall(), ShortfallCause::InsideWorkdir));
}
let reason = match declared {
ContainmentClass::Unconfined => {
"unconfined declares no floor and is met by every session".to_string()
}
ContainmentClass::Advisory => {
"advisory is the weakest containment class and is satisfied by every host".to_string()
}
ContainmentClass::Scoped => {
"scoped requires Landlock filesystem mediation (Linux 5.13+ with a usable Landlock \
ABI); this host provides no kernel filesystem mediation, so paths outside the \
workdir are constrained by convention only"
.to_string()
}
ContainmentClass::Sealed => match sealed_shortfall {
Some(shortfall) => shortfall.reason(),
None => "sealed requires a private mount namespace pivoted onto a composed root \
(unshare(CLONE_NEWUSER|CLONE_NEWNS) + pivot_root), which this host did not \
provide"
.to_string(),
},
};
Some((reason, ShortfallCause::Host))
}
const WORKDIR_EXEC_SHORTFALL: &str =
"capabilities.filesystem.workdir_exec: true keeps the Landlock Execute right on the session \
workdir, so a binary the capsule compiles, downloads or renames inside it runs regardless of \
capabilities.shell.allow — the allowlist stops being an enforceable property of this capsule. \
No host can back a class above advisory for it. Either remove workdir_exec (the allowlist is \
then enforced by the kernel on the resolved path) or lower the declared containment floor to \
advisory";
pub fn check_containment_floor(
declared: ContainmentClass,
achieved: ContainmentClass,
sealed_shortfall: Option<&SealedShortfall>,
workdir_exec: bool,
runtime_dir_inside: Option<&RuntimeDirPlacement>,
) -> Result<(), RuntimeError> {
match containment_shortfall(
declared,
achieved,
sealed_shortfall,
workdir_exec,
runtime_dir_inside,
) {
None => Ok(()),
Some((reason, cause)) => Err(RuntimeError::ContainmentFloorUnmet {
declared,
achieved,
reason,
cause,
required_by: Vec::new(),
}),
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExportsFilesReport {
pub root: String,
pub mode: ExportMode,
pub max_bytes: u64,
}
impl From<&FileExport> for ExportsFilesReport {
fn from(export: &FileExport) -> Self {
Self {
root: export.root.clone(),
mode: export.mode,
max_bytes: export.max_bytes,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct PeerFilesReport {
pub root: String,
pub max_ttl_secs: u64,
pub max_bytes: u64,
}
impl From<&PeerFilesExport> for PeerFilesReport {
fn from(export: &PeerFilesExport) -> Self {
Self {
root: export.root.clone(),
max_ttl_secs: export.effective_max_ttl_secs(),
max_bytes: export.max_bytes,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct StateStoreReport {
pub artifact: String,
pub store: String,
pub host_path: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum PreopenSurface {
WholeWorkdir,
ScopedSubtree,
Nothing,
RecordsHooks,
}
impl PreopenSurface {
fn summary(self) -> &'static str {
match self {
PreopenSurface::WholeWorkdir => "the whole accessible workdir",
PreopenSurface::ScopedSubtree => "one subtree of the accessible workdir",
PreopenSurface::Nothing => "nothing preopened",
PreopenSurface::RecordsHooks => "the records directory's hooks/",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct PreopenReport {
pub artifact: String,
pub role: String,
pub scope: Option<String>,
pub surface: PreopenSurface,
pub runtime_dir_reach: Option<RuntimeDirReach>,
pub extra_roots: Vec<ExtraRootReport>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExtraRootReport {
pub guest_path: String,
pub path: String,
pub access: crate::host_paths::HostPathAccess,
pub source: crate::guest_roots::GuestRootSource,
}
impl From<&crate::guest_roots::GuestRoot> for ExtraRootReport {
fn from(root: &crate::guest_roots::GuestRoot) -> Self {
Self {
guest_path: root.guest_path.clone(),
path: root.host_path.display().to_string(),
access: root.access,
source: root.source,
}
}
}
pub fn attach_extra_roots(reports: &mut [PreopenReport], roots: &[crate::guest_roots::GuestRoot]) {
for report in reports {
report.extra_roots = if report.runtime_dir_reach.is_some() {
roots.iter().map(ExtraRootReport::from).collect()
} else {
Vec::new()
};
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum RuntimeDirReach {
MurmurRuntime,
}
impl PreopenReport {
pub fn render(&self) -> String {
let artifact = &self.artifact;
let role = &self.role;
let surface = self.surface.summary();
let guest = crate::runtime_dir::RUNTIME_DIR_GUEST_PATH;
let runtime_dir = match self.runtime_dir_reach {
None => String::new(),
Some(RuntimeDirReach::MurmurRuntime) => {
format!("; runtime directory read-only at {guest}/")
}
};
let extra_roots = self.render_extra_roots();
match &self.scope {
Some(scope) => format!(
"{artifact} ({role}): {surface} — capabilities.filesystem.scope: \
{scope}{runtime_dir}{extra_roots}"
),
None => format!(
"{artifact} ({role}): {surface} — no capabilities.filesystem.scope \
declared{runtime_dir}{extra_roots}"
),
}
}
fn render_extra_roots(&self) -> String {
if self.extra_roots.is_empty() {
return String::new();
}
let guest = crate::guest_roots::HOST_ROOT_GUEST_PATH;
let whole_host = self
.extra_roots
.iter()
.any(|root| root.source == crate::guest_roots::GuestRootSource::Unconfined);
if whole_host {
return format!("; whole host read-write under {guest}/");
}
let folders: Vec<String> = self
.extra_roots
.iter()
.map(|root| format!("{} ({})", root.path, root.access))
.collect();
format!("; host folders under {guest}/: {}", folders.join(", "))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum FilesystemRestriction {
Advisory,
Enforced,
Absent,
None,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct FilesystemBoundaryReport {
pub restriction: FilesystemRestriction,
pub not_protected: Vec<String>,
}
const ADVISORY_FILESYSTEM_NOT_PROTECTED: &str =
"Filesystem restriction is advisory on this host: no kernel mechanism denies a read or a write \
outside the work folder, so capabilities.filesystem.scope bounds what the runtime hands a \
subprocess rather than what that subprocess can reach. A Linux host with a usable Landlock \
ABI installs a Landlock domain instead, where the kernel refuses the open.";
const ENFORCED_FILESYSTEM_NOT_PROTECTED: &str =
"A path outside the granted set is denied rather than absent: Landlock refuses the open, and \
stat, access and readlink on that path still succeed and still report the real file's \
metadata, because Landlock mediates the operations that touch a file and not path resolution. \
A probe that only asks whether a path exists therefore learns nothing about what was denied — \
read the file to see the refusal. Declaring capabilities.containment: sealed on a host that \
reaches the sealed tier composes a root the path is not in at all.";
const ADVISORY_HOME_NOT_PROTECTED: &str =
"HOME is relocated, not protected: it points at the session's home folder, and the real home \
directory stays readable and writable by absolute path, because no kernel mechanism mediates \
the filesystem on this tier.";
const PATH_REWRITING_NOT_PROTECTED: &str =
"Path rewriting is a convenience, not a boundary: HOME is set to the session's home folder, \
home/ in its records directory, on every tier, so `cat ~/.ssh/id_rsa` answers \"No such file \
or directory\" whether or not anything was denied — the tilde resolved into that folder and \
the host file was never looked for. A ~-based probe is therefore never evidence of containment; test with an absolute path \
to a real file outside the workdir instead.";
pub(crate) const UNCONFINED_NOT_PROTECTED: &str =
"Nothing is contained: this session runs unconfined, so its shell commands, plan shell steps \
and process harness run as the operator, with the operator's HOME, environment \
(credential variables included), filesystem and network, and sudo works wherever the \
operator's sudoers allows it. The trace and the records directory are written as always but \
are not protected from those commands.";
pub(crate) fn filesystem_boundary(applied: EnforcementTier) -> FilesystemBoundaryReport {
if applied == EnforcementTier::Unconfined {
return FilesystemBoundaryReport {
restriction: FilesystemRestriction::None,
not_protected: vec![UNCONFINED_NOT_PROTECTED.to_string()],
};
}
let not_protected: &[&str] = match applied {
EnforcementTier::KernelSealed => &[],
EnforcementTier::Unconfined => unreachable!("returned above"),
EnforcementTier::KernelFull => &[
ENFORCED_FILESYSTEM_NOT_PROTECTED,
PATH_REWRITING_NOT_PROTECTED,
],
EnforcementTier::KernelSeccompOnly | EnforcementTier::EnvironmentOnly => &[
ADVISORY_FILESYSTEM_NOT_PROTECTED,
PATH_REWRITING_NOT_PROTECTED,
ADVISORY_HOME_NOT_PROTECTED,
],
};
let restriction = match applied {
EnforcementTier::KernelSealed => FilesystemRestriction::Absent,
EnforcementTier::Unconfined => unreachable!("returned above"),
EnforcementTier::KernelFull => FilesystemRestriction::Enforced,
EnforcementTier::KernelSeccompOnly | EnforcementTier::EnvironmentOnly => {
FilesystemRestriction::Advisory
}
};
let not_protected = not_protected
.iter()
.map(|statement| (*statement).to_string())
.collect();
FilesystemBoundaryReport {
restriction,
not_protected,
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct RuntimeDirReport {
pub path: String,
pub read_only_to_commands: bool,
pub enforced_by: Option<&'static str>,
pub not_protected_because: Option<String>,
}
const RUNTIME_DIR_ENFORCED_SEALED: &str = "mountns+landlock";
const RUNTIME_DIR_ENFORCED_LANDLOCK: &str = "landlock";
const LANDLOCK_TRUNCATE_ABI: u8 = 3;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProtectedDir {
Runtime,
Records,
}
impl ProtectedDir {
fn noun(self) -> &'static str {
match self {
Self::Runtime => "runtime directory",
Self::Records => "records directory",
}
}
fn file(self) -> &'static str {
match self {
Self::Runtime => "runtime file",
Self::Records => "record",
}
}
fn heading(self) -> &'static str {
match self {
Self::Runtime => "Runtime directory",
Self::Records => "Records directory",
}
}
}
#[must_use]
pub(crate) fn runtime_dir_report(
applied: EnforcementTier,
landlock_abi: Option<u8>,
placement: &RuntimeDirPlacement,
) -> RuntimeDirReport {
protected_dir_report(
ProtectedDir::Runtime,
applied,
landlock_abi,
placement.report_path(),
placement.sessions_inside_workdir,
&placement.workdir,
)
}
#[must_use]
pub(crate) fn records_dir_report(
applied: EnforcementTier,
landlock_abi: Option<u8>,
placement: &RuntimeDirPlacement,
) -> RuntimeDirReport {
protected_dir_report(
ProtectedDir::Records,
applied,
landlock_abi,
placement.records_report_path(),
placement.records_inside_workdir,
&placement.workdir,
)
}
fn protected_dir_report(
dir: ProtectedDir,
applied: EnforcementTier,
landlock_abi: Option<u8>,
path: String,
inside_workdir: bool,
workdir: &std::path::Path,
) -> RuntimeDirReport {
let noun = dir.noun();
let not_protected = |reason: String| RuntimeDirReport {
path: path.clone(),
read_only_to_commands: false,
enforced_by: None,
not_protected_because: Some(reason),
};
match applied {
EnforcementTier::Unconfined => not_protected(format!(
"the capsule runs unconfined, so its commands run as the operator and can change the \
{noun}"
)),
EnforcementTier::EnvironmentOnly => not_protected(format!(
"this host has no kernel containment for capsule commands — macOS and every \
non-Linux host run them as the operator's own processes, so a command can change \
the {noun}"
)),
EnforcementTier::KernelSeccompOnly => not_protected(format!(
"this host has no Landlock, so no kernel mechanism keeps a command from changing the \
{noun}"
)),
_ if inside_workdir => not_protected(format!(
"the {noun} {path} lies inside the workdir {}, so the capsule's commands can write \
it; run with a --workdir that does not contain ~/.murmur",
workdir.display()
)),
EnforcementTier::KernelSealed => RuntimeDirReport {
path,
read_only_to_commands: true,
enforced_by: Some(RUNTIME_DIR_ENFORCED_SEALED),
not_protected_because: None,
},
EnforcementTier::KernelFull => {
let abi = landlock_abi.unwrap_or(1);
if abi < LANDLOCK_TRUNCATE_ABI {
not_protected(format!(
"Landlock ABI {abi} does not mediate truncate(2), so a command can empty a \
{} though not replace it; kernel 6.2 or later refuses it",
dir.file()
))
} else {
RuntimeDirReport {
path,
read_only_to_commands: true,
enforced_by: Some(RUNTIME_DIR_ENFORCED_LANDLOCK),
not_protected_because: None,
}
}
}
}
}
#[must_use]
pub fn render_runtime_dir(report: &RuntimeDirReport) -> String {
render_protected_dir(ProtectedDir::Runtime, report)
}
#[must_use]
pub fn render_records_dir(report: &RuntimeDirReport) -> String {
render_protected_dir(ProtectedDir::Records, report)
}
fn render_protected_dir(dir: ProtectedDir, report: &RuntimeDirReport) -> String {
let commands = match (&report.not_protected_because, report.enforced_by) {
(Some(reason), _) => format!("can change it — {reason}"),
(None, Some(mechanism)) => format!("read only ({mechanism})"),
(None, None) => "read only".to_string(),
};
format!(
"{}\n path: {}\n commands: {commands}\n",
dir.heading(),
report.path
)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct TempDirReport {
pub path: String,
pub tmpdir: String,
pub tmp_mapped: bool,
pub mapped_by: Option<&'static str>,
pub not_mapped_because: Option<String>,
}
#[must_use]
pub(crate) fn temp_dir_report(
mapping: crate::temp_dir::TempMapping,
placement: &RuntimeDirPlacement,
) -> TempDirReport {
let path = std::path::Path::new(&placement.records_report_path())
.join(crate::temp_dir::SESSION_TEMP_DIR_NAME)
.display()
.to_string();
TempDirReport {
tmpdir: mapping.tmpdir_value(std::path::Path::new(&path)),
path,
tmp_mapped: mapping.is_mapped(),
mapped_by: mapping.mapped_by(),
not_mapped_because: match mapping {
crate::temp_dir::TempMapping::NotMapped(reason) => Some(reason.describe().to_string()),
_ => None,
},
}
}
#[must_use]
pub fn render_temp_dir(report: &TempDirReport) -> String {
let tmp = match (report.mapped_by, &report.not_mapped_because) {
(Some(mechanism), _) => format!("mapped ({mechanism})"),
(None, Some(reason)) => format!("not mapped — {reason}"),
(None, None) => "not mapped".to_string(),
};
format!(
"Temporary folder\n path: {}\n TMPDIR: {}\n /tmp: {tmp}\n",
report.path, report.tmpdir
)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ScopeReport {
pub declared_containment: ContainmentClass,
pub achieved_containment: ContainmentClass,
pub host_containment: ContainmentClass,
pub floor_met: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub shortfall_reason: Option<String>,
pub enforcement_tier: &'static str,
pub routes: Vec<RouteReport>,
pub userns_grant: Option<UsernsGrant>,
pub landlock_abi: Option<u8>,
pub filesystem_scope: Option<String>,
pub workdir_exec: bool,
pub read_only_paths: Vec<String>,
pub read_only_advisory_for: Vec<String>,
pub preopens: Vec<PreopenReport>,
pub network_allow: Vec<String>,
pub unix_sockets: bool,
pub shell_allow: Vec<String>,
pub spawn_allow: Vec<String>,
pub spawn_workdir_snapshot: Vec<String>,
pub parent_workdir: Option<crate::workdir_snapshot::ParentWorkdirReport>,
pub images: Option<ImagesReport>,
pub env_allow: Vec<String>,
pub plan_submit: bool,
pub install_skill: Vec<String>,
pub install_tool: Vec<String>,
pub install_from: Vec<String>,
pub interpreter_runtime_grants: Vec<String>,
pub derived_interpreter_runtime: Vec<String>,
pub trust_store: Vec<String>,
pub name_resolution: Vec<String>,
pub host_paths: Vec<crate::host_paths::HostPathReport>,
pub exports_files: Option<ExportsFilesReport>,
pub peer_files: Option<PeerFilesReport>,
pub peer_tasks: bool,
pub peer_fetch_allow: Vec<String>,
pub state_stores: Vec<StateStoreReport>,
pub configured_artifacts: Vec<String>,
pub staged_runtime_grants: Vec<String>,
pub runtime_writes: Vec<crate::workdir_writes::RuntimeWriteReport>,
pub manifest_files: Vec<crate::manifest_files::ManifestFileReport>,
pub io_max: IoMaxReport,
pub cgroup_scope: CgroupScopeReport,
pub filesystem_boundary: FilesystemBoundaryReport,
pub runtime_dir: RuntimeDirReport,
pub records_dir: RuntimeDirReport,
pub temp_dir: TempDirReport,
pub process_view: crate::proc_view::ProcessViewReport,
pub unconfined: Option<UnconfinedReport>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct UnconfinedReport {
pub user: Option<String>,
pub uid: u32,
pub home: Option<String>,
pub applies_to: Vec<&'static str>,
pub resource_limits: Vec<&'static str>,
}
pub const UNCONFINED_APPLIES_TO: &[&str] = &["shell", "plan-shell-step", "process-harness"];
impl UnconfinedReport {
pub(crate) fn for_this_process(declared: &murmur_artifact::ResourceCapabilities) -> Self {
let uid = current_uid();
Self {
user: account_name(uid),
uid,
home: std::env::var_os("HOME").map(|home| home.to_string_lossy().into_owned()),
applies_to: UNCONFINED_APPLIES_TO.to_vec(),
resource_limits: crate::resources::declared_resource_fields(declared),
}
}
#[must_use]
pub fn render(&self) -> String {
let applies_to = self
.applies_to
.iter()
.map(|route| match *route {
"shell" => "shell commands",
"plan-shell-step" => "plan shell steps",
"process-harness" => "process harness",
other => other,
})
.collect::<Vec<_>>()
.join(", ");
let limits = if self.resource_limits.is_empty() {
"none declared".to_string()
} else {
self.resource_limits.join(", ")
};
format!(
"Unconfined\n user: {}\n uid: {}\n HOME: {}\n applies to: \
{applies_to}\n resource limits: {limits}\n",
self.user.as_deref().unwrap_or("<unknown>"),
self.uid,
self.home.as_deref().unwrap_or("<unset>"),
)
}
}
pub(crate) fn unconfined_warning(report: &UnconfinedReport) -> String {
let user = match &report.user {
Some(user) => format!("{user} (uid {})", report.uid),
None => format!("uid {}", report.uid),
};
let home = report.home.as_deref().unwrap_or("<unset>");
format!(
"this capsule runs unconfined: its shell commands, plan shell steps and process harness \
run as {user} with HOME {home}, the whole environment mur run was \
started with (credential variables included), the host filesystem and network as that \
account has them, and no no_new_privs, so sudo works wherever that account's sudoers \
allows it"
)
}
pub(crate) fn warn_unconfined_launch(
records_root: &crate::workdir_fs::WorkdirRoot,
report: &UnconfinedReport,
) {
let code = murmur_artifact::security_warnings::W_SEC_036;
let message = unconfined_warning(report);
let link = murmur_artifact::security_warning_link(code);
crate::runtime_err!("[capsule-runtime] warning[{code}]: {message} ({link})");
crate::agent::append_bootstrap_log(
records_root,
&format!("[capability-policy] warning[{code}]: {message} ({link})"),
);
}
#[cfg(unix)]
pub(crate) fn current_uid() -> u32 {
rustix::process::getuid().as_raw()
}
#[cfg(not(unix))]
pub(crate) fn current_uid() -> u32 {
0
}
#[cfg(unix)]
#[allow(unsafe_code)]
pub(crate) fn account_name(uid: u32) -> Option<String> {
let mut buffer = vec![0 as libc::c_char; 16 * 1024];
let mut entry: libc::passwd = unsafe { std::mem::zeroed() };
let mut result: *mut libc::passwd = std::ptr::null_mut();
let rc = unsafe {
libc::getpwuid_r(
uid,
&mut entry,
buffer.as_mut_ptr(),
buffer.len(),
&mut result,
)
};
if rc != 0 || result.is_null() || entry.pw_name.is_null() {
return None;
}
let name = unsafe { std::ffi::CStr::from_ptr(entry.pw_name) };
Some(name.to_string_lossy().into_owned())
}
#[cfg(not(unix))]
pub(crate) fn account_name(_uid: u32) -> Option<String> {
None
}
impl ScopeReport {
pub fn render(&self) -> String {
let mut out = String::new();
out.push_str("Containment\n");
out.push_str(&format!(" declared: {}\n", self.declared_containment));
out.push_str(&format!(" achieved: {}\n", self.achieved_containment));
out.push_str(&format!(" host can back: {}\n", self.host_containment));
out.push_str(&format!(
" floor met: {}\n",
if self.floor_met { "yes" } else { "no" }
));
if let Some(reason) = &self.shortfall_reason {
out.push_str(&format!(" reason: {reason}\n"));
}
out.push_str(&format!(" mechanism: {}\n", self.enforcement_tier));
out.push_str(&format!(
" userns grant: {}\n",
match self.userns_grant {
Some(grant) => grant.wire_name(),
None => "n/a",
}
));
out.push_str(&format!(
" landlock ABI: {}\n",
match self.landlock_abi {
Some(abi) => abi.to_string(),
None => "n/a".to_string(),
}
));
out.push_str(&format!(
" {}\n",
crate::cgroup::render_cgroup_scope(&self.cgroup_scope)
));
out.push('\n');
out.push_str(&render_routes(&self.routes));
if let Some(unconfined) = &self.unconfined {
out.push('\n');
out.push_str(&unconfined.render());
}
out.push('\n');
out.push_str(&render_runtime_dir(&self.runtime_dir));
out.push('\n');
out.push_str(&render_records_dir(&self.records_dir));
out.push('\n');
out.push_str(&render_temp_dir(&self.temp_dir));
out.push('\n');
out.push_str(&crate::proc_view::render_process_view(&self.process_view));
out.push_str(&render_not_protected(&self.filesystem_boundary));
out.push_str("\nEffective grants\n");
out.push_str(&format!(
" filesystem scope: {}\n",
self.filesystem_scope.as_deref().unwrap_or("<none>")
));
out.push_str(&format!(" workdir exec: {}\n", self.workdir_exec));
out.push_str(&render_read_only(
&self.read_only_paths,
&self.read_only_advisory_for,
));
push_list(
&mut out,
"preopens",
&self
.preopens
.iter()
.map(PreopenReport::render)
.collect::<Vec<_>>(),
);
push_list(&mut out, "network allow", &self.network_allow);
out.push_str(&format!(" unix sockets: {}\n", self.unix_sockets));
push_list(&mut out, "shell allow", &self.shell_allow);
push_list(&mut out, "spawn allow", &self.spawn_allow);
push_list(
&mut out,
"spawn workdir snapshot",
&self.spawn_workdir_snapshot,
);
push_list(&mut out, "env allow", &self.env_allow);
out.push_str(&format!(" plan submit: {}\n", self.plan_submit));
out.push_str(&render_install(
&self.install_skill,
&self.install_tool,
&self.install_from,
));
push_list(
&mut out,
"interpreter runtime",
&self.interpreter_runtime_grants,
);
push_list(
&mut out,
"derived interpreter runtime",
&self.derived_interpreter_runtime,
);
push_list(&mut out, "trust store", &self.trust_store);
push_list(&mut out, "name resolution", &self.name_resolution);
push_list(
&mut out,
"host paths",
&self
.host_paths
.iter()
.map(crate::host_paths::HostPathReport::render)
.collect::<Vec<_>>(),
);
push_list(&mut out, "staged runtime", &self.staged_runtime_grants);
push_list(
&mut out,
"state stores",
&self
.state_stores
.iter()
.map(|store| format!("{}: {} -> {}", store.artifact, store.store, store.host_path))
.collect::<Vec<_>>(),
);
push_list(&mut out, "artifact config", &self.configured_artifacts);
if let Some(parent) = &self.parent_workdir {
out.push_str(&render_parent_workdir(parent));
}
if let Some(images) = &self.images {
out.push_str(&render_images(images));
}
out.push_str("\nResource plane\n");
match &self.exports_files {
None => out.push_str(" exports.files: <none>\n"),
Some(export) => {
out.push_str(&format!(" exports.files root: {}\n", export.root));
out.push_str(&format!(" mode: {}\n", export.mode));
out.push_str(&format!(
" max bytes: {} (per file)\n",
export.max_bytes
));
}
}
out.push_str(&format!(
" io.max: {} ({} bytes/s declared)\n",
match self.io_max.status {
IoMaxStatus::Enforced => "enforced",
IoMaxStatus::Unavailable => "unavailable",
IoMaxStatus::NotRequired => "not-required",
IoMaxStatus::NotProbed => "not-probed",
},
self.io_max.declared_bytes_per_sec
));
if let Some(reason) = &self.io_max.reason {
out.push_str(&format!(" {reason}\n"));
}
out.push_str("\nPeer handoff\n");
match &self.peer_files {
None => out.push_str(" exports.peer_files: <none>\n"),
Some(peer) => {
out.push_str(&format!(" exports.peer_files root: {}\n", peer.root));
out.push_str(&format!(
" max handle ttl: {}s\n",
peer.max_ttl_secs
));
out.push_str(&format!(
" max bytes: {} (per file)\n",
peer.max_bytes
));
}
}
out.push_str(if self.peer_tasks {
" exports.peer_tasks: accept\n"
} else {
" exports.peer_tasks: refuse\n"
});
push_list(&mut out, "peer_fetch allow", &self.peer_fetch_allow);
out.push_str("\nRuntime writes\n");
out.push_str(&crate::workdir_writes::render_runtime_writes(
&self.runtime_writes,
));
out.push_str(&crate::manifest_files::render_manifest_files(
&self.manifest_files,
));
if !self.floor_met {
out.push_str(
"\nThis is a report only — `mur run` without --explain-scope would refuse to \
launch here.\n",
);
} else if self.cgroup_scope.status == CgroupScopeStatus::Unavailable {
out.push_str(
"\nThis is a report only — `mur run` without --explain-scope would refuse to \
launch here with E-RUN-012: this capsule can run a subprocess and no \
cgroup scope can be created to bound it (see `cgroup scope:` above).\n",
);
}
out
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct ImagesReport {
pub keep: u32,
pub max_bytes: u64,
pub max_per_turn: u32,
pub keep_applies: bool,
}
impl ScopeReport {
#[must_use]
pub fn with_images(
mut self,
images: murmur_artifact::ContextImagesConfig,
transport: Option<&str>,
) -> Self {
self.images = Some(ImagesReport {
keep: images.keep,
max_bytes: images.max_bytes,
max_per_turn: images.max_per_turn,
keep_applies: transport != Some("process"),
});
self
}
#[must_use]
pub fn with_routes(mut self, transport: Option<&str>) -> Self {
let inputs = RouteInputs::new(
&self.shell_allow,
self.plan_submit,
&self.install_tool,
&self.preopens,
transport,
);
self.routes = route_table(&inputs, self.enforcement_tier);
self
}
#[must_use]
pub fn with_manifest_files(
mut self,
rows: Vec<crate::manifest_files::ManifestFileReport>,
) -> Self {
self.manifest_files = rows;
self
}
#[must_use]
pub fn with_process_view(mut self, view: &crate::proc_view::ProcView) -> Self {
self.process_view = crate::proc_view::process_view_report(view);
self
}
#[must_use]
pub fn with_cgroup_scope(mut self, cgroup_scope: CgroupScopeReport) -> Self {
self.cgroup_scope = cgroup_scope;
self
}
#[must_use]
pub fn with_parent_workdir(
mut self,
snapshot: Option<crate::workdir_snapshot::WorkdirSnapshot>,
) -> Self {
let restriction = self.filesystem_boundary.restriction;
self.parent_workdir =
snapshot.map(|snapshot| crate::workdir_snapshot::ParentWorkdirReport {
snapshot,
restriction,
});
self
}
}
#[must_use]
pub fn render_images(images: &ImagesReport) -> String {
let mut out = String::from("\nImages\n");
out.push_str(&format!(" max_bytes: {}\n", images.max_bytes));
if images.keep_applies {
out.push_str(&format!(" keep: {}\n", images.keep));
out.push_str(&format!(" max_per_turn: {}\n", images.max_per_turn));
} else {
out.push_str(&format!(
" keep: {} (does nothing here: the harness holds the conversation)\n",
images.keep
));
out.push_str(&format!(
" max_per_turn: {} (counts the images of one tool call here)\n",
images.max_per_turn
));
}
out
}
#[must_use]
pub fn render_parent_workdir(parent: &crate::workdir_snapshot::ParentWorkdirReport) -> String {
let snapshot = &parent.snapshot;
let mut out = String::from("\nParent work folder\n");
out.push_str(&format!(" source: {}\n", snapshot.source));
out.push_str(&format!(
" copied: {} files, {} directories, {} symlinks ({} re-pointed), {} bytes\n",
snapshot.files,
snapshot.dirs,
snapshot.symlinks,
snapshot.repointed_symlinks,
snapshot.bytes
));
if snapshot.skipped_count == 0 {
out.push_str(" skipped: <none>\n");
} else {
out.push_str(&format!(" skipped: {}\n", snapshot.skipped_count));
for entry in &snapshot.skipped {
let reason = serde_json::to_value(entry.reason)
.ok()
.and_then(|value| value.as_str().map(str::to_string))
.unwrap_or_default();
out.push_str(&format!(" - {} ({reason})\n", entry.path));
}
}
out.push_str(&format!(
" restriction: {}\n",
parent.restriction_sentence()
));
out
}
#[must_use]
pub fn read_only_advisory_for(read_only_paths: &[String], shell_allow: &[String]) -> Vec<String> {
if read_only_paths.is_empty() {
return Vec::new();
}
shell_allow
.iter()
.filter(|binary| crate::protected_paths::is_advisory_interpreter(binary))
.cloned()
.collect()
}
#[must_use]
pub fn render_read_only(read_only_paths: &[String], advisory_for: &[String]) -> String {
let mut out = String::new();
push_list(&mut out, "read_only", read_only_paths);
if read_only_paths.is_empty() {
return out;
}
if advisory_for.is_empty() {
out.push_str(
" read_only enforcement: enforced for every tool call and every shell command the \
dispatch check can read\n",
);
} else {
out.push_str(&format!(
" read_only enforcement: advisory against {} — an interpreter's own file I/O is not \
something the dispatch check can read; enforced for every tool call and for every \
shell command it can read\n",
advisory_for.join(", ")
));
}
out
}
#[must_use]
pub fn render_install(
install_skill: &[String],
install_tool: &[String],
install_from: &[String],
) -> String {
let mut out = String::new();
push_list(&mut out, "install skill", install_skill);
push_list(&mut out, "install tool", install_tool);
if !install_from.is_empty() {
out.push_str(&format!(" install from: {}\n", install_from.join(", ")));
}
out
}
#[must_use]
pub fn install_from_places(policy: &CapabilityPolicy) -> Vec<String> {
if !crate::install_grant::is_granted(policy) {
return Vec::new();
}
let mut places = vec!["stores".to_string()];
if policy.install_from_sources {
places.push("sources".to_string());
}
places
}
fn render_not_protected(boundary: &FilesystemBoundaryReport) -> String {
if boundary.not_protected.is_empty() {
return String::new();
}
let mut out = String::from("\nNot protected here\n");
for statement in &boundary.not_protected {
out.push_str(&format!(" - {statement}\n"));
}
out
}
fn push_list(out: &mut String, label: &str, values: &[String]) {
if values.is_empty() {
out.push_str(&format!(" {label}: <none>\n"));
return;
}
out.push_str(&format!(" {label}:\n"));
for value in values {
out.push_str(&format!(" - {value}\n"));
}
}
#[allow(clippy::too_many_arguments)]
pub fn explain_scope(
policy: &CapabilityPolicy,
declared: ContainmentClass,
exports: Option<&Exports>,
state_stores: Vec<StateStoreReport>,
configured_artifacts: Vec<String>,
preopens: Vec<PreopenReport>,
io_max: IoMaxReport,
workdir: &std::path::Path,
) -> Result<ScopeReport, crate::murmur_home::HomeUnusable> {
let placement = crate::runtime_dir::runtime_dir_placement(workdir)?;
let probe = HostProbe::probe();
let decision = ContainmentDecision::decide(
probe.tier(),
declared,
policy.workdir_exec_allowed,
placement.inside_workdir(),
);
let temp_mapping = crate::temp_dir::temp_mapping(
decision.applied_tier(),
probe.mount_namespace_usable(),
&placement.workdir,
);
Ok(scope_report_for_tier(
policy,
&decision,
probe.sealed_shortfall().as_ref(),
probe.userns_grant(),
probe.landlock_kernel_abi(),
exports,
state_stores,
configured_artifacts,
preopens,
&placement,
temp_mapping,
io_max,
)
.with_process_view(&crate::proc_view::proc_view(
decision.applied_tier(),
probe.namespaces(),
)))
}
pub fn doctor_runtime_dir(
probe: &HostProbe,
declared: ContainmentClass,
workdir: &std::path::Path,
) -> Result<DoctorDirReports, crate::murmur_home::HomeUnusable> {
let placement = crate::runtime_dir::runtime_dir_placement(workdir)?;
let applied = ContainmentDecision::decide(probe.tier(), declared, false, false).applied_tier();
let abi = crate::sandbox::ruleset_landlock_abi(probe.landlock_kernel_abi(), applied);
let temp_mapping =
crate::temp_dir::temp_mapping(applied, probe.mount_namespace_usable(), &placement.workdir);
Ok(DoctorDirReports {
runtime_dir: runtime_dir_report(applied, abi, &placement),
records_dir: records_dir_report(applied, abi, &placement),
temp_dir: temp_dir_report(temp_mapping, &placement),
process_view: crate::proc_view::process_view_report(&crate::proc_view::proc_view(
applied,
probe.namespaces(),
)),
placement,
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DoctorDirReports {
pub runtime_dir: RuntimeDirReport,
pub records_dir: RuntimeDirReport,
pub temp_dir: TempDirReport,
pub process_view: crate::proc_view::ProcessViewReport,
pub placement: RuntimeDirPlacement,
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn scope_report_for_tier(
policy: &CapabilityPolicy,
decision: &ContainmentDecision,
sealed_shortfall: Option<&SealedShortfall>,
userns_grant: Option<UsernsGrant>,
landlock_kernel_abi: Option<u8>,
exports: Option<&Exports>,
state_stores: Vec<StateStoreReport>,
configured_artifacts: Vec<String>,
preopens: Vec<PreopenReport>,
runtime_dir_placement: &RuntimeDirPlacement,
temp_mapping: crate::temp_dir::TempMapping,
io_max: IoMaxReport,
) -> ScopeReport {
debug_assert_eq!(
*decision,
ContainmentDecision::decide(
decision.host_tier,
decision.declared,
policy.workdir_exec_allowed,
runtime_dir_placement.inside_workdir(),
),
"the decision must be made from the policy and placement this report states"
);
let declared = decision.declared();
let achieved = decision.achieved();
let applied = decision.applied_tier();
let shortfall_reason = containment_shortfall_reason(
declared,
achieved,
sealed_shortfall,
policy.workdir_exec_allowed,
runtime_dir_placement.inside(),
);
let landlock_abi = crate::sandbox::ruleset_landlock_abi(landlock_kernel_abi, applied);
let runtime_dir = runtime_dir_report(applied, landlock_abi, runtime_dir_placement);
let records_dir = records_dir_report(applied, landlock_abi, runtime_dir_placement);
let mut filesystem_boundary = filesystem_boundary(applied);
let enforcement_tier = decision.enforcement_tier();
let routes = route_table(
&RouteInputs::new(
&policy.shell_allow,
policy.plan_submit,
&policy.install_tool,
&preopens,
None,
),
enforcement_tier,
);
if applied != EnforcementTier::Unconfined {
for report in [&runtime_dir, &records_dir] {
if let Some(reason) = &report.not_protected_because {
filesystem_boundary.not_protected.push(reason.clone());
}
}
}
let unconfined = (applied == EnforcementTier::Unconfined)
.then(|| UnconfinedReport::for_this_process(&policy.declared_resources));
ScopeReport {
declared_containment: declared,
achieved_containment: achieved,
host_containment: decision.host_containment(),
floor_met: shortfall_reason.is_none(),
shortfall_reason,
enforcement_tier,
routes,
userns_grant,
landlock_abi,
exports_files: exports
.and_then(|exports| exports.files.as_ref())
.map(ExportsFilesReport::from),
peer_files: exports
.and_then(|exports| exports.peer_files.as_ref())
.map(PeerFilesReport::from),
peer_tasks: exports.is_some_and(Exports::accepts_peer_tasks),
peer_fetch_allow: policy.peer_fetch_allow.clone(),
state_stores,
configured_artifacts,
filesystem_scope: policy.filesystem_scope.clone(),
workdir_exec: policy.workdir_exec_allowed,
read_only_paths: policy.read_only_paths.clone(),
read_only_advisory_for: read_only_advisory_for(
&policy.read_only_paths,
&policy.shell_allow,
),
preopens,
network_allow: policy.network_allow.clone(),
unix_sockets: policy.unix_sockets_allowed,
shell_allow: policy.shell_allow.clone(),
spawn_allow: policy.spawn_allow.clone(),
spawn_workdir_snapshot: policy.spawn_workdir_snapshot.clone(),
parent_workdir: None,
images: None,
env_allow: policy.env_allow.clone(),
plan_submit: policy.plan_submit,
install_skill: policy.install_skill.clone(),
install_tool: policy.install_tool.clone(),
install_from: install_from_places(policy),
interpreter_runtime_grants: policy
.shell_interpreter_runtime
.iter()
.flat_map(|grant| {
grant.dirs.iter().map(move |dir| {
let listable = if dir.list_dir { " (list_dir)" } else { "" };
format!("{}: {}{}", grant.binary, dir.path, listable)
})
})
.collect(),
derived_interpreter_runtime: policy
.shell_derived_interpreter_runtime
.iter()
.flat_map(|runtime| {
let binary = runtime.binary.display();
runtime
.entries()
.map(|(path, is_dir)| {
let listable = if is_dir { " (list_dir)" } else { "" };
format!("{binary}: {}{listable}", path.display())
})
.chain(
runtime
.libraries
.iter()
.map(|library| format!("{binary}: {}", library.display())),
)
.collect::<Vec<_>>()
})
.collect(),
name_resolution: match applied {
EnforcementTier::KernelFull | EnforcementTier::KernelSealed => {
crate::sandbox::SCOPED_NAME_RESOLUTION_PATHS
.iter()
.map(|path| (*path).to_string())
.collect()
}
_ => Vec::new(),
},
trust_store: policy
.shell_trust_store
.iter()
.map(|entry| {
let listable = if entry.directory { " (list_dir)" } else { "" };
format!("{}{listable}", entry.path.display())
})
.collect(),
host_paths: crate::host_paths::host_path_reports(
&policy.host_paths,
crate::host_paths::HostPathMechanism::for_enforcement_tier(decision.enforcement_tier()),
),
staged_runtime_grants: policy
.shell_staged_runtime
.iter()
.map(|grant| {
format!(
"{}: {} (pin: {})",
grant.binary, grant.source_path, grant.pin
)
})
.collect(),
runtime_writes: crate::workdir_writes::runtime_writes(),
manifest_files: Vec::new(),
io_max,
cgroup_scope: CgroupScopeReport::default(),
filesystem_boundary,
runtime_dir,
records_dir,
temp_dir: temp_dir_report(temp_mapping, runtime_dir_placement),
process_view: crate::proc_view::process_view_report(&crate::proc_view::tier_proc_view(
applied,
)),
unconfined,
}
}
#[cfg(test)]
#[allow(clippy::too_many_arguments)]
pub(crate) fn report_for(
policy: &CapabilityPolicy,
declared: ContainmentClass,
tier: EnforcementTier,
sealed_shortfall: Option<&SealedShortfall>,
userns_grant: Option<UsernsGrant>,
landlock_kernel_abi: Option<u8>,
exports: Option<&Exports>,
state_stores: Vec<StateStoreReport>,
configured_artifacts: Vec<String>,
preopens: Vec<PreopenReport>,
runtime_dir_placement: &RuntimeDirPlacement,
temp_mapping: crate::temp_dir::TempMapping,
io_max: IoMaxReport,
) -> ScopeReport {
let decision = ContainmentDecision::decide(
tier,
declared,
policy.workdir_exec_allowed,
runtime_dir_placement.inside_workdir(),
);
scope_report_for_tier(
policy,
&decision,
sealed_shortfall,
userns_grant,
landlock_kernel_abi,
exports,
state_stores,
configured_artifacts,
preopens,
runtime_dir_placement,
temp_mapping,
io_max,
)
}
pub const HOST_PROCESS_MECHANISM: &str = "host-process";
const UNCONFINED_ROUTE_NOTE: &str = "Runs as the operator, with their HOME, whole environment, \
filesystem, network and sudo; nothing holds it.";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum Route {
Shell,
PlanShellSteps,
WasmTools,
Drivers,
Hooks,
ProcessHarness,
HarnessTools,
}
impl Route {
pub const ALL: [Self; 7] = [
Self::Shell,
Self::PlanShellSteps,
Self::WasmTools,
Self::Drivers,
Self::Hooks,
Self::ProcessHarness,
Self::HarnessTools,
];
#[must_use]
pub fn label(self) -> &'static str {
match self {
Self::Shell => "shell commands",
Self::PlanShellSteps => "plan shell steps",
Self::WasmTools => "WASM tools",
Self::Drivers => "drivers",
Self::Hooks => "hooks",
Self::ProcessHarness => "the process harness",
Self::HarnessTools => "harness tools",
}
}
#[must_use]
pub fn under_class(self) -> bool {
match self {
Self::Shell | Self::PlanShellSteps => true,
Self::WasmTools
| Self::Drivers
| Self::Hooks
| Self::ProcessHarness
| Self::HarnessTools => false,
}
}
#[must_use]
pub fn runs_unconfined(self) -> bool {
match self {
Self::Shell | Self::PlanShellSteps | Self::ProcessHarness => true,
Self::WasmTools | Self::Drivers | Self::Hooks | Self::HarnessTools => false,
}
}
#[must_use]
pub fn note(self) -> Option<&'static str> {
match self {
Self::Shell | Self::PlanShellSteps => None,
Self::WasmTools => Some(
"Runs inside the mur process, held by its WASI preopens and the host-side network \
allowlist only.",
),
Self::Drivers => Some(
"Runs inside the mur process, held by its WASI preopens and the host-side network \
allowlist only.",
),
Self::Hooks => Some(
"Runs inside the mur process, held only by the preopens and network grants its \
own capabilities block declares.",
),
Self::ProcessHarness => Some(
"A child of mur with a cleared environment and nothing else: no Landlock, \
seccomp, network namespace or rlimits, in a cgroup with no limits.",
),
Self::HarnessTools => Some(
"Run by the harness with the user's authority, kept off only by the arguments \
the process driver passes.",
),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct RouteReport {
pub route: Route,
pub present: bool,
pub under_class: bool,
pub mechanism: &'static str,
pub note: Option<&'static str>,
pub artifacts: Vec<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RouteInputs {
pub shell_allow: bool,
pub plan_submit: bool,
pub install_tool: bool,
pub tools: Vec<String>,
pub drivers: Vec<String>,
pub hooks: Vec<String>,
pub process_transport: bool,
}
impl RouteInputs {
#[must_use]
pub fn new(
shell_allow: &[String],
plan_submit: bool,
install_tool: &[String],
preopens: &[PreopenReport],
transport: Option<&str>,
) -> Self {
let named = |role: murmur_artifact::ArtifactRuntime| {
preopens
.iter()
.filter(|preopen| preopen.role == role.as_str())
.map(|preopen| preopen.artifact.clone())
.collect::<Vec<_>>()
};
Self {
shell_allow: !shell_allow.is_empty(),
plan_submit,
install_tool: !install_tool.is_empty(),
tools: named(murmur_artifact::ArtifactRuntime::Tool),
drivers: named(murmur_artifact::ArtifactRuntime::Driver),
hooks: named(murmur_artifact::ArtifactRuntime::Hook),
process_transport: transport == Some("process"),
}
}
}
#[must_use]
pub fn route_table(inputs: &RouteInputs, enforcement_tier: &'static str) -> Vec<RouteReport> {
let unconfined = enforcement_tier == enforcement_tier_name(EnforcementTier::Unconfined);
Route::ALL
.into_iter()
.map(|route| {
let (present, artifacts) = match route {
Route::Shell => (inputs.shell_allow, Vec::new()),
Route::PlanShellSteps => (inputs.plan_submit && inputs.shell_allow, Vec::new()),
Route::WasmTools => (
!inputs.tools.is_empty() || inputs.install_tool,
inputs.tools.clone(),
),
Route::Drivers => (!inputs.drivers.is_empty(), inputs.drivers.clone()),
Route::Hooks => (!inputs.hooks.is_empty(), inputs.hooks.clone()),
Route::ProcessHarness | Route::HarnessTools => {
(inputs.process_transport, Vec::new())
}
};
if unconfined && route.runs_unconfined() {
return RouteReport {
route,
present,
under_class: true,
mechanism: enforcement_tier,
note: Some(UNCONFINED_ROUTE_NOTE),
artifacts,
};
}
let under_class = route.under_class();
RouteReport {
route,
present,
under_class,
mechanism: if under_class {
enforcement_tier
} else {
HOST_PROCESS_MECHANISM
},
note: route.note(),
artifacts,
}
})
.collect()
}
#[must_use]
pub fn render_routes(routes: &[RouteReport]) -> String {
let mut out = String::from("Routes\n");
for row in routes {
let label = match row.route {
Route::ProcessHarness => "process harness",
route => route.label(),
};
let presence = if row.present { "present" } else { "absent" };
let mechanism = if row.under_class {
row.mechanism
} else {
"host process, not confined by the class"
};
let artifacts = if row.artifacts.is_empty() {
String::new()
} else {
format!(" ({})", row.artifacts.join(", "))
};
out.push_str(&format!(
" {label:<17} {presence:<8} {mechanism}{artifacts}\n"
));
if let (true, Some(note)) = (row.present, row.note) {
out.push_str(&format!(" {note}\n"));
}
}
out
}
#[must_use]
pub fn uncovered_route_warning(
declared: ContainmentClass,
routes: &[RouteReport],
) -> Option<String> {
if declared < ContainmentClass::Scoped {
return None;
}
let uncovered: Vec<String> = routes
.iter()
.filter(|row| row.present && !row.under_class)
.map(|row| {
if row.artifacts.is_empty() {
row.route.label().to_string()
} else {
format!("{} ({})", row.route.label(), row.artifacts.join(", "))
}
})
.collect();
let named = match uncovered.as_slice() {
[] => return None,
[only] => only.clone(),
[rest @ .., last] => format!("{} and {last}", rest.join(", ")),
};
Some(format!(
"the declared {declared} floor does not confine every route this capsule runs code on: \
{named} run outside the class, with none of its Landlock, seccomp or network namespace \
around them (`mur run --explain-scope` lists what holds each under Routes). The class \
confines shell commands and plan shell steps; nothing is refused"
))
}
pub fn warn_on_uncovered_routes(
declared: ContainmentClass,
routes: &[RouteReport],
bootstrap_log: Option<&crate::workdir_fs::WorkdirRoot>,
) {
let Some(message) = uncovered_route_warning(declared, routes) else {
return;
};
let code = murmur_artifact::W_SEC_034;
let link = murmur_artifact::security_warning_link(code);
crate::runtime_err!("[capsule-runtime] warning[{code}]: {message} ({link})");
if let Some(records) = bootstrap_log {
crate::agent::append_bootstrap_log(
records,
&format!("[capability-policy] warning[{code}]: {message} ({link})"),
);
}
}
pub fn applied_mechanism(probe: &HostProbe, declared: ContainmentClass) -> &'static str {
ContainmentDecision::decide(probe.tier(), declared, false, false).enforcement_tier()
}
fn enforcement_tier_name(tier: EnforcementTier) -> &'static str {
match tier {
EnforcementTier::KernelSealed => "mountns+pivot_root+landlock+seccomp",
EnforcementTier::KernelFull => "landlock+seccomp",
EnforcementTier::KernelSeccompOnly => "seccomp-only",
EnforcementTier::EnvironmentOnly => "none",
EnforcementTier::Unconfined => "unconfined",
}
}
#[cfg(test)]
#[allow(clippy::print_stdout, clippy::print_stderr)]
mod tests {
use super::*;
use crate::sealed::SealedBlocker;
fn blocked(blocker: SealedBlocker) -> SealedShortfall {
SealedShortfall {
blocker,
failed_call: None,
errno: None,
userns_grant: UsernsGrant::Withheld,
attachment: crate::sealed::ProfileAttachment::NotReported,
}
}
use murmur_artifact::{InterpreterRuntimeDir, InterpreterRuntimeGrant, StagedRuntimeGrant};
const ALL_TIERS: &[EnforcementTier] = &[
EnforcementTier::KernelSealed,
EnforcementTier::KernelFull,
EnforcementTier::KernelSeccompOnly,
EnforcementTier::EnvironmentOnly,
];
#[test]
fn kernel_full_achieves_scoped() {
assert_eq!(
achieved_class_for_tier(EnforcementTier::KernelFull),
ContainmentClass::Scoped
);
}
#[test]
fn only_the_sealed_tier_achieves_sealed() {
assert_eq!(
achieved_class_for_tier(EnforcementTier::KernelSealed),
ContainmentClass::Sealed
);
for tier in ALL_TIERS
.iter()
.filter(|t| **t != EnforcementTier::KernelSealed)
{
assert_ne!(achieved_class_for_tier(*tier), ContainmentClass::Sealed);
}
}
#[test]
fn tiers_without_landlock_achieve_only_advisory() {
assert_eq!(
achieved_class_for_tier(EnforcementTier::KernelSeccompOnly),
ContainmentClass::Advisory
);
assert_eq!(
achieved_class_for_tier(EnforcementTier::EnvironmentOnly),
ContainmentClass::Advisory
);
}
#[test]
fn workdir_exec_caps_the_achieved_class_at_advisory_on_every_tier() {
assert_eq!(
achieved_containment_class(EnforcementTier::KernelSealed, false, false),
ContainmentClass::Sealed
);
assert_eq!(
achieved_containment_class(EnforcementTier::KernelFull, false, false),
ContainmentClass::Scoped
);
assert_eq!(
achieved_containment_class(EnforcementTier::KernelSeccompOnly, false, false),
ContainmentClass::Advisory
);
assert_eq!(
achieved_containment_class(EnforcementTier::EnvironmentOnly, false, false),
ContainmentClass::Advisory
);
for tier in ALL_TIERS {
assert_eq!(
achieved_containment_class(*tier, true, false),
ContainmentClass::Advisory,
"workdir_exec must cap tier {tier:?} at advisory"
);
}
}
#[test]
fn the_tier_only_mapping_is_unaffected_by_the_cap() {
for tier in ALL_TIERS {
assert_eq!(
achieved_class_for_tier(*tier),
achieved_containment_class(*tier, false, false),
"the tier-only mapping and the uncapped wrapper must agree on tier {tier:?}"
);
}
}
#[test]
fn workdir_exec_refuses_a_scoped_floor_even_on_a_landlock_capable_host() {
let achieved = achieved_containment_class(EnforcementTier::KernelFull, true, false);
assert_eq!(achieved, ContainmentClass::Advisory);
let error = check_containment_floor(ContainmentClass::Scoped, achieved, None, true, None)
.expect_err("workdir_exec + scoped must refuse");
match error {
RuntimeError::ContainmentFloorUnmet {
declared,
achieved,
reason,
..
} => {
assert_eq!(declared, ContainmentClass::Scoped);
assert_eq!(achieved, ContainmentClass::Advisory);
assert!(
reason.contains("workdir_exec"),
"the reason must name the manifest key, got: {reason}"
);
assert!(
!reason.contains("this host provides no kernel filesystem mediation"),
"the reason must not blame the host for a manifest declaration: {reason}"
);
}
other => panic!("expected ContainmentFloorUnmet, got {other:?}"),
}
}
#[test]
fn workdir_exec_outranks_a_probed_sealed_blocker_in_the_reason() {
let reason = containment_shortfall_reason(
ContainmentClass::Sealed,
ContainmentClass::Advisory,
Some(&blocked(SealedBlocker::AppArmorProfileMissing)),
true,
None,
)
.expect("refusal");
assert!(reason.contains("workdir_exec"));
assert!(!reason.contains("apparmor_parser"));
}
#[test]
fn workdir_exec_alone_refuses_nothing() {
for tier in ALL_TIERS {
let achieved = achieved_containment_class(*tier, true, false);
assert!(
check_containment_floor(ContainmentClass::Advisory, achieved, None, true, None)
.is_ok(),
"workdir_exec at an advisory floor must launch on tier {tier:?}"
);
}
}
#[test]
fn advisory_floor_is_met_on_every_tier() {
for tier in ALL_TIERS {
let achieved = achieved_class_for_tier(*tier);
assert!(check_containment_floor(
ContainmentClass::Advisory,
achieved,
None,
false,
None
)
.is_ok());
}
}
#[test]
fn scoped_floor_is_met_on_every_landlock_capable_tier() {
for tier in [EnforcementTier::KernelFull, EnforcementTier::KernelSealed] {
let achieved = achieved_class_for_tier(tier);
assert!(
check_containment_floor(ContainmentClass::Scoped, achieved, None, false, None)
.is_ok()
);
}
for tier in [
EnforcementTier::KernelSeccompOnly,
EnforcementTier::EnvironmentOnly,
] {
let achieved = achieved_class_for_tier(tier);
let error =
check_containment_floor(ContainmentClass::Scoped, achieved, None, false, None)
.expect_err("scoped must refuse without Landlock");
match error {
RuntimeError::ContainmentFloorUnmet {
declared,
achieved,
reason,
..
} => {
assert_eq!(declared, ContainmentClass::Scoped);
assert_eq!(achieved, ContainmentClass::Advisory);
assert!(
reason.contains("Landlock"),
"reason must name the missing mechanism, got: {reason}"
);
}
other => panic!("expected ContainmentFloorUnmet, got {other:?}"),
}
}
}
#[test]
fn a_weaker_declaration_is_satisfied_by_a_sealed_host_without_being_upgraded() {
for declared in [ContainmentClass::Advisory, ContainmentClass::Scoped] {
let decision =
ContainmentDecision::decide(EnforcementTier::KernelSealed, declared, false, false);
assert_eq!(decision.applied_tier(), EnforcementTier::KernelFull);
assert_eq!(decision.achieved(), ContainmentClass::Scoped);
assert_eq!(decision.enforcement_tier(), "landlock+seccomp");
assert_eq!(decision.host_containment(), ContainmentClass::Sealed);
assert!(
check_containment_floor(declared, decision.achieved(), None, false, None).is_ok()
);
assert_eq!(
containment_shortfall_reason(declared, decision.achieved(), None, false, None),
None
);
}
}
const ALL_CLASSES: &[ContainmentClass] = &[
ContainmentClass::Advisory,
ContainmentClass::Scoped,
ContainmentClass::Sealed,
];
#[test]
fn the_decision_reads_the_class_off_the_tier_the_session_installs() {
use ContainmentClass::{Advisory, Scoped, Sealed};
use EnforcementTier::{EnvironmentOnly, KernelFull, KernelSealed, KernelSeccompOnly};
let table = [
(
(KernelSealed, Advisory),
(KernelFull, Scoped, "landlock+seccomp"),
),
(
(KernelSealed, Scoped),
(KernelFull, Scoped, "landlock+seccomp"),
),
(
(KernelSealed, Sealed),
(KernelSealed, Sealed, "mountns+pivot_root+landlock+seccomp"),
),
(
(KernelFull, Advisory),
(KernelFull, Scoped, "landlock+seccomp"),
),
(
(KernelFull, Scoped),
(KernelFull, Scoped, "landlock+seccomp"),
),
(
(KernelFull, Sealed),
(KernelFull, Scoped, "landlock+seccomp"),
),
(
(KernelSeccompOnly, Advisory),
(KernelSeccompOnly, Advisory, "seccomp-only"),
),
(
(KernelSeccompOnly, Scoped),
(KernelSeccompOnly, Advisory, "seccomp-only"),
),
(
(KernelSeccompOnly, Sealed),
(KernelSeccompOnly, Advisory, "seccomp-only"),
),
(
(EnvironmentOnly, Advisory),
(EnvironmentOnly, Advisory, "none"),
),
(
(EnvironmentOnly, Scoped),
(EnvironmentOnly, Advisory, "none"),
),
(
(EnvironmentOnly, Sealed),
(EnvironmentOnly, Advisory, "none"),
),
];
assert_eq!(table.len(), ALL_TIERS.len() * ALL_CLASSES.len());
for ((host_tier, declared), (applied, class, wire)) in table {
let host_class = match host_tier {
KernelSealed => Sealed,
KernelFull => Scoped,
KernelSeccompOnly | EnvironmentOnly => Advisory,
EnforcementTier::Unconfined => unreachable!("never a host tier"),
};
for workdir_exec in [false, true] {
for inside in [false, true] {
let decision =
ContainmentDecision::decide(host_tier, declared, workdir_exec, inside);
let what = format!("{host_tier:?} {declared} exec={workdir_exec} in={inside}");
assert_eq!(decision.declared(), declared, "{what}");
assert_eq!(decision.applied_tier(), applied, "{what}");
assert_eq!(decision.enforcement_tier(), wire, "{what}");
assert_eq!(decision.host_containment(), host_class, "{what}");
let expected = if workdir_exec || inside {
Advisory
} else {
class
};
assert_eq!(decision.achieved(), expected, "{what}");
assert!(decision.achieved() <= decision.host_containment(), "{what}");
assert!(
decision.achieved() != Sealed || declared == Sealed,
"{what}"
);
}
}
}
let scoped = ContainmentDecision::decide(KernelSealed, Scoped, false, false);
assert_eq!(
(
scoped.achieved(),
scoped.enforcement_tier(),
scoped.host_containment()
),
(Scoped, "landlock+seccomp", Sealed)
);
let sealed = ContainmentDecision::decide(KernelSealed, Sealed, false, false);
assert_eq!(
(sealed.achieved(), sealed.enforcement_tier()),
(Sealed, "mountns+pivot_root+landlock+seccomp")
);
let advisory = ContainmentDecision::decide(KernelSealed, Advisory, false, false);
assert_eq!(
(advisory.achieved(), advisory.enforcement_tier()),
(Scoped, "landlock+seccomp")
);
}
#[test]
fn the_floor_verdict_and_wording_are_those_of_the_host_tier_class() {
for &host_tier in ALL_TIERS {
for &declared in ALL_CLASSES {
for workdir_exec in [false, true] {
for inside in [None, Some(inside_placement())] {
let in_workdir = inside.is_some();
let decision = ContainmentDecision::decide(
host_tier,
declared,
workdir_exec,
in_workdir,
);
let host_class =
achieved_containment_class(host_tier, workdir_exec, in_workdir);
let shortfall = blocked(SealedBlocker::NamespaceCreationDenied);
assert_eq!(
containment_shortfall_reason(
declared,
decision.achieved(),
Some(&shortfall),
workdir_exec,
inside.as_ref(),
),
containment_shortfall_reason(
declared,
host_class,
Some(&shortfall),
workdir_exec,
inside.as_ref(),
),
"{host_tier:?} {declared} exec={workdir_exec} in={in_workdir}"
);
}
}
}
}
}
#[test]
fn sealed_floor_is_met_only_on_the_sealed_tier() {
assert!(check_containment_floor(
ContainmentClass::Sealed,
achieved_class_for_tier(EnforcementTier::KernelSealed),
None,
false,
None
)
.is_ok());
for tier in ALL_TIERS
.iter()
.filter(|t| **t != EnforcementTier::KernelSealed)
{
let achieved = achieved_class_for_tier(*tier);
let error = check_containment_floor(
ContainmentClass::Sealed,
achieved,
Some(&blocked(SealedBlocker::NamespaceCreationDenied)),
false,
None,
)
.expect_err("sealed must refuse on a host that cannot back it");
match error {
RuntimeError::ContainmentFloorUnmet {
declared, reason, ..
} => {
assert_eq!(declared, ContainmentClass::Sealed);
assert!(
reason.contains("seccomp profile") && !reason.contains("--cap-add"),
"reason must name the blocking mechanism's remediation, got: {reason}"
);
}
other => panic!("expected ContainmentFloorUnmet, got {other:?}"),
}
}
}
#[test]
fn the_sealed_refusal_names_the_specific_blocker() {
let achieved = ContainmentClass::Scoped;
let apparmor = containment_shortfall_reason(
ContainmentClass::Sealed,
achieved,
Some(&blocked(SealedBlocker::AppArmorProfileMissing)),
false,
None,
)
.expect("refusal");
assert!(apparmor.contains("mur-sealed"));
assert!(apparmor.contains("apparmor_parser -r"));
assert!(!apparmor.contains("--cap-add"));
let container = containment_shortfall_reason(
ContainmentClass::Sealed,
achieved,
Some(&blocked(SealedBlocker::NamespaceCreationDenied)),
false,
None,
)
.expect("refusal");
assert!(!container.contains("--cap-add"), "{container}");
assert!(container.contains("seccomp profile"), "{container}");
assert!(container.contains("unshare(CLONE_NEWUSER)"), "{container}");
assert!(container.contains("--security-opt seccomp="), "{container}");
assert!(
container.contains("seccompProfile: Localhost"),
"{container}"
);
assert!(container.contains("outside the container"));
assert!(!container.contains("apparmor_parser"));
let pivot = containment_shortfall_reason(
ContainmentClass::Sealed,
ContainmentClass::Advisory,
Some(&blocked(SealedBlocker::PivotRootDenied)),
false,
None,
)
.expect("refusal");
assert!(pivot.contains("pivot_root(2)"), "reason was: {pivot}");
assert_ne!(
pivot,
containment_shortfall_reason(
ContainmentClass::Sealed,
ContainmentClass::Advisory,
Some(&blocked(SealedBlocker::MountDenied)),
false,
None,
)
.expect("refusal")
);
for blocker in [
SealedBlocker::NotLinux,
SealedBlocker::AppArmorProfileMissing,
SealedBlocker::NamespaceCreationDenied,
SealedBlocker::MountDenied,
SealedBlocker::PivotRootDenied,
SealedBlocker::KernelUnsupported,
SealedBlocker::LandlockUnavailable,
] {
let reason = containment_shortfall_reason(
ContainmentClass::Sealed,
achieved,
Some(&blocked(blocker)),
false,
None,
)
.expect("refusal");
assert!(
!reason.contains("no host can provide it today"),
"got: {reason}"
);
}
}
#[test]
fn the_sealed_refusal_falls_back_to_naming_the_mechanism_without_a_probe() {
let reason = containment_shortfall_reason(
ContainmentClass::Sealed,
ContainmentClass::Advisory,
None,
false,
None,
)
.expect("refusal");
assert!(reason.contains("pivot_root"));
assert!(reason.contains("CLONE_NEWNS"));
}
#[test]
fn shortfall_reason_is_absent_when_the_floor_is_met() {
assert_eq!(
containment_shortfall_reason(
ContainmentClass::Advisory,
ContainmentClass::Scoped,
None,
false,
None
),
None
);
assert_eq!(
containment_shortfall_reason(
ContainmentClass::Scoped,
ContainmentClass::Scoped,
None,
false,
None
),
None
);
assert_eq!(
containment_shortfall_reason(
ContainmentClass::Sealed,
ContainmentClass::Sealed,
Some(&blocked(SealedBlocker::NamespaceCreationDenied)),
false,
None
),
None,
"a met floor must ignore a stale blocker rather than manufacture a refusal"
);
}
fn sample_policy() -> CapabilityPolicy {
CapabilityPolicy {
network_allow: vec!["https://api.example.com".to_string()],
unix_sockets_allowed: false,
filesystem_scope: Some("workdir".to_string()),
shell_allow: vec!["python3".to_string()],
spawn_allow: vec!["helper".to_string()],
env_allow: vec!["TZ".to_string()],
shell_interpreter_runtime: vec![InterpreterRuntimeGrant {
binary: "python3".to_string(),
dirs: vec![InterpreterRuntimeDir {
path: "/usr/lib/python3.11".to_string(),
list_dir: true,
}],
}],
..CapabilityPolicy::default()
}
}
#[test]
fn an_unconfined_report_names_no_protection_it_does_not_install() {
for host_tier in [
EnforcementTier::KernelSealed,
EnforcementTier::KernelFull,
EnforcementTier::KernelSeccompOnly,
EnforcementTier::EnvironmentOnly,
] {
let decision =
ContainmentDecision::decide(host_tier, ContainmentClass::Unconfined, true, true);
assert_eq!(decision.applied_tier(), EnforcementTier::Unconfined);
assert_eq!(decision.achieved(), ContainmentClass::Unconfined);
assert_eq!(decision.enforcement_tier(), "unconfined");
assert_eq!(
decision.host_containment(),
achieved_class_for_tier(host_tier)
);
}
let policy = CapabilityPolicy {
containment_floor: ContainmentClass::Unconfined,
declared_resources: murmur_artifact::ResourceCapabilities {
max_open_files: Some(256),
..Default::default()
},
..sample_policy()
};
let report = report_for(
&policy,
ContainmentClass::Unconfined,
EnforcementTier::KernelFull,
None,
None,
Some(4),
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(EnforcementTier::Unconfined),
IoMaxReport::default(),
);
assert!(report.floor_met);
assert_eq!(report.achieved_containment, ContainmentClass::Unconfined);
assert_eq!(report.enforcement_tier, "unconfined");
for row in &report.routes {
if row.route.runs_unconfined() {
assert!(row.under_class, "{row:?}");
assert_eq!(row.mechanism, "unconfined", "{row:?}");
assert_eq!(row.note, Some(UNCONFINED_ROUTE_NOTE), "{row:?}");
} else {
assert_eq!(row.note, row.route.note(), "{row:?}");
}
}
assert_eq!(
uncovered_route_warning(report.declared_containment, &report.routes),
None,
"W-SEC-034 keys on a scoped or sealed floor"
);
assert_eq!(report.landlock_abi, None);
assert_eq!(
report.filesystem_boundary.restriction,
FilesystemRestriction::None
);
assert_eq!(
report.filesystem_boundary.not_protected,
vec![UNCONFINED_NOT_PROTECTED.to_string()]
);
for dir in [&report.runtime_dir, &report.records_dir] {
assert!(!dir.read_only_to_commands);
assert_eq!(dir.enforced_by, None);
assert!(dir
.not_protected_because
.as_deref()
.unwrap()
.contains("unconfined"));
}
assert!(!report.temp_dir.tmp_mapped);
assert_eq!(
report.temp_dir.not_mapped_because.as_deref(),
Some(crate::temp_dir::NotMappedReason::Unconfined.describe())
);
let unconfined = report.unconfined.as_ref().expect("an unconfined report");
assert_eq!(unconfined.uid, current_uid());
assert_eq!(unconfined.applies_to, UNCONFINED_APPLIES_TO.to_vec());
assert_eq!(unconfined.resource_limits, vec!["max_open_files"]);
let json = serde_json::to_value(&report).unwrap();
assert_eq!(json["enforcement_tier"], "unconfined");
assert_eq!(json["achieved_containment"], "unconfined");
assert_eq!(json["filesystem_boundary"]["restriction"], "none");
assert!(json["unconfined"].is_object(), "{json}");
let rendered = report.render();
assert!(rendered.contains(" achieved: unconfined\n"), "{rendered}");
assert!(rendered.contains("\nUnconfined\n user:"), "{rendered}");
assert!(rendered.contains(" uid: "), "{rendered}");
assert!(rendered.contains(" HOME: "), "{rendered}");
assert!(
rendered.contains("applies to: shell commands, plan shell steps, process harness"),
"{rendered}"
);
assert!(unconfined_warning(unconfined).contains("sudo"));
let contained = report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(EnforcementTier::KernelFull),
IoMaxReport::default(),
);
let json = serde_json::to_value(&contained).unwrap();
assert!(json["unconfined"].is_null(), "{json}");
assert!(json.as_object().unwrap().contains_key("unconfined"));
}
#[test]
fn scope_report_mirrors_the_policy_and_the_probed_tier() {
let report = report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
assert_eq!(report.declared_containment, ContainmentClass::Scoped);
assert_eq!(report.achieved_containment, ContainmentClass::Scoped);
assert!(report.floor_met);
assert_eq!(report.shortfall_reason, None);
assert_eq!(report.enforcement_tier, "landlock+seccomp");
assert_eq!(report.filesystem_scope.as_deref(), Some("workdir"));
assert_eq!(report.network_allow, vec!["https://api.example.com"]);
assert!(!report.unix_sockets);
assert_eq!(report.shell_allow, vec!["python3"]);
assert_eq!(report.spawn_allow, vec!["helper"]);
assert_eq!(report.env_allow, vec!["TZ"]);
assert!(!report.workdir_exec);
assert_eq!(
report.interpreter_runtime_grants,
vec!["python3: /usr/lib/python3.11 (list_dir)"]
);
}
#[test]
fn scope_report_surfaces_workdir_exec_and_the_advisory_it_forces() {
let policy = CapabilityPolicy {
shell_allow: vec!["gcc".to_string()],
workdir_exec_allowed: true,
..CapabilityPolicy::default()
};
let report = report_for(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
assert!(report.workdir_exec);
assert_eq!(report.achieved_containment, ContainmentClass::Advisory);
assert!(!report.floor_met);
assert_eq!(report.enforcement_tier, "landlock+seccomp");
assert!(report
.shortfall_reason
.as_deref()
.expect("refusal")
.contains("workdir_exec"));
let value: serde_json::Value = serde_json::to_value(&report).unwrap();
assert_eq!(value["workdir_exec"], true);
let rendered = report.render();
assert!(rendered.contains("workdir exec: true"));
assert!(rendered.contains("achieved: advisory"));
}
#[test]
fn scope_report_json_always_carries_io_max() {
let unprobed: serde_json::Value = serde_json::to_value(report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
))
.unwrap();
assert_eq!(unprobed["io_max"]["status"], "not-probed");
assert_eq!(unprobed["io_max"]["declared_bytes_per_sec"], 0);
assert!(unprobed["io_max"].get("reason").is_none());
let enforced: serde_json::Value = serde_json::to_value(report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport {
declared_bytes_per_sec: 52_428_800,
status: IoMaxStatus::Enforced,
reason: None,
},
))
.unwrap();
assert_eq!(enforced["io_max"]["status"], "enforced");
assert_eq!(enforced["io_max"]["declared_bytes_per_sec"], 52_428_800);
}
#[test]
fn the_rendered_resource_plane_states_what_became_of_the_io_ceiling() {
let rendered = report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport {
declared_bytes_per_sec: 52_428_800,
status: IoMaxStatus::Unavailable,
reason: Some("the io controller is not delegated".to_string()),
},
)
.render();
assert!(rendered.contains("Resource plane"), "{rendered}");
assert!(rendered.contains("io.max:"), "{rendered}");
assert!(rendered.contains("unavailable"), "{rendered}");
assert!(rendered.contains("52428800 bytes/s declared"), "{rendered}");
assert!(
rendered.contains("the io controller is not delegated"),
"{rendered}"
);
}
#[test]
fn scope_report_json_always_carries_workdir_exec() {
let value: serde_json::Value = serde_json::to_value(report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
))
.unwrap();
assert_eq!(value["workdir_exec"], false);
}
#[test]
fn staged_runtime_grants_are_reported_on_every_tier() {
let policy = CapabilityPolicy {
shell_allow: vec!["python3".to_string()],
shell_staged_runtime: vec![StagedRuntimeGrant {
binary: "python3".to_string(),
source_path: "/opt/testbed/conda/envs/django__django".to_string(),
pin: "conda-4.10.3/python-3.9.19".to_string(),
}],
..CapabilityPolicy::default()
};
for tier in ALL_TIERS {
let report = report_for(
&policy,
ContainmentClass::Sealed,
*tier,
Some(&blocked(SealedBlocker::NamespaceCreationDenied)),
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
*tier,
ContainmentClass::Sealed,
)),
IoMaxReport::default(),
);
assert_eq!(
report.staged_runtime_grants,
vec![
"python3: /opt/testbed/conda/envs/django__django \
(pin: conda-4.10.3/python-3.9.19)"
],
"tier {tier:?} must still report the declared staged runtime"
);
assert!(
report.render().contains("staged runtime"),
"the rendered report must carry a staged runtime section on tier {tier:?}"
);
}
}
#[test]
fn absent_staged_runtime_reports_nothing() {
let report = report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
assert!(report.staged_runtime_grants.is_empty());
assert!(report.render().contains("staged runtime: <none>"));
}
#[test]
fn grant_set_never_raises_the_achieved_class() {
let report = report_for(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::KernelFull,
Some(&blocked(SealedBlocker::NamespaceCreationDenied)),
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Sealed,
)),
IoMaxReport::default(),
);
assert_eq!(report.achieved_containment, ContainmentClass::Scoped);
assert!(!report.floor_met);
}
#[test]
fn scope_report_still_reports_when_the_floor_is_unmet() {
let report = report_for(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::EnvironmentOnly,
Some(&blocked(SealedBlocker::NotLinux)),
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::EnvironmentOnly,
ContainmentClass::Sealed,
)),
IoMaxReport::default(),
);
assert_eq!(report.declared_containment, ContainmentClass::Sealed);
assert_eq!(report.achieved_containment, ContainmentClass::Advisory);
assert!(!report.floor_met);
assert!(report.shortfall_reason.unwrap().contains("non-Linux"));
assert_eq!(report.enforcement_tier, "none");
}
#[test]
fn scope_report_names_the_sealed_mechanism_on_a_sealed_host() {
let report = report_for(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::KernelSealed,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSealed,
ContainmentClass::Sealed,
)),
IoMaxReport::default(),
);
assert_eq!(report.achieved_containment, ContainmentClass::Sealed);
assert_eq!(report.host_containment, ContainmentClass::Sealed);
assert!(report.floor_met);
assert_eq!(
report.enforcement_tier,
"mountns+pivot_root+landlock+seccomp"
);
assert_eq!(
report.filesystem_boundary.restriction,
FilesystemRestriction::Absent
);
}
#[test]
fn every_report_field_follows_the_decision() {
for &host_tier in ALL_TIERS {
for &declared in ALL_CLASSES {
let decision = ContainmentDecision::decide(host_tier, declared, false, false);
let report = scope_report_for_tier(
&sample_policy(),
&decision,
None,
None,
Some(6),
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(decision.applied_tier()),
IoMaxReport::default(),
);
let what = format!("{host_tier:?} {declared}");
assert_eq!(report.declared_containment, declared, "{what}");
assert_eq!(report.achieved_containment, decision.achieved(), "{what}");
assert_eq!(
report.enforcement_tier,
decision.enforcement_tier(),
"{what}"
);
assert_eq!(
report.host_containment,
decision.host_containment(),
"{what}"
);
assert_eq!(
report.landlock_abi,
crate::sandbox::ruleset_landlock_abi(Some(6), decision.applied_tier()),
"{what}"
);
let restriction = filesystem_boundary(decision.applied_tier()).restriction;
assert_eq!(
report.filesystem_boundary.restriction, restriction,
"{what}"
);
assert_eq!(
restriction,
match report.enforcement_tier {
"mountns+pivot_root+landlock+seccomp" => FilesystemRestriction::Absent,
"landlock+seccomp" => FilesystemRestriction::Enforced,
_ => FilesystemRestriction::Advisory,
},
"{what}"
);
assert_eq!(
report.temp_dir,
temp_dir_report(
crate::temp_dir::tier_temp_mapping(decision.applied_tier()),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
),
"{what}"
);
}
}
}
#[test]
fn host_containment_is_reported_beside_the_achieved_class() {
let report = report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelSealed,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(EnforcementTier::KernelFull),
IoMaxReport::default(),
);
let value = serde_json::to_value(&report).unwrap();
assert_eq!(value["achieved_containment"], "scoped");
assert_eq!(value["host_containment"], "sealed");
assert_eq!(value["enforcement_tier"], "landlock+seccomp");
assert_eq!(value["filesystem_boundary"]["restriction"], "enforced");
let rendered = report.render();
assert!(
rendered.contains(" achieved: scoped\n host can back: sealed\n floor met: yes\n"),
"{rendered}"
);
assert!(
rendered.contains(" mechanism: landlock+seccomp\n"),
"{rendered}"
);
}
#[test]
fn scope_report_json_uses_wire_names() {
let report = report_for(
&CapabilityPolicy::default(),
ContainmentClass::Advisory,
EnforcementTier::KernelSeccompOnly,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSeccompOnly,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
);
let value: serde_json::Value = serde_json::to_value(&report).unwrap();
assert_eq!(value["declared_containment"], "advisory");
assert_eq!(value["achieved_containment"], "advisory");
assert_eq!(value["host_containment"], "advisory");
assert_eq!(value["floor_met"], true);
assert_eq!(value["enforcement_tier"], "seccomp-only");
assert!(value.get("shortfall_reason").is_none());
}
#[test]
fn scope_report_names_the_landlock_abi_the_ruleset_is_built_at() {
let report = |tier, kernel| {
report_for(
&sample_policy(),
ContainmentClass::Scoped,
tier,
None,
None,
kernel,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
tier,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
)
};
for (tier, kernel, expected) in [
(EnforcementTier::KernelFull, Some(8), Some(3)),
(EnforcementTier::KernelSealed, Some(8), Some(3)),
(EnforcementTier::KernelFull, Some(2), Some(2)),
(EnforcementTier::KernelFull, Some(1), Some(1)),
(EnforcementTier::KernelFull, None, None),
(EnforcementTier::KernelSeccompOnly, Some(8), None),
(EnforcementTier::EnvironmentOnly, None, None),
] {
let scope = report(tier, kernel);
assert_eq!(
scope.landlock_abi, expected,
"{tier:?} with kernel {kernel:?}"
);
let value: serde_json::Value = serde_json::to_value(&scope).unwrap();
assert!(value.as_object().unwrap().contains_key("landlock_abi"));
match expected {
Some(abi) => assert_eq!(value["landlock_abi"], abi),
None => assert!(value["landlock_abi"].is_null()),
}
let rendered = scope.render();
let line = match expected {
Some(abi) => format!(" landlock ABI: {abi}\n"),
None => " landlock ABI: n/a\n".to_string(),
};
assert!(rendered.contains(&line), "{rendered}");
assert!(
rendered.find("userns grant:").unwrap() < rendered.find("landlock ABI:").unwrap()
);
}
}
#[test]
fn the_userns_grant_is_reported_without_changing_any_verdict() {
let baseline = report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSealed,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
);
for grant in UsernsGrant::ALL {
let report = report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
Some(*grant),
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSealed,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
);
assert_eq!(report.achieved_containment, baseline.achieved_containment);
assert_eq!(report.floor_met, baseline.floor_met);
assert_eq!(report.enforcement_tier, baseline.enforcement_tier);
assert_eq!(report.userns_grant, Some(*grant));
let value: serde_json::Value = serde_json::to_value(&report).unwrap();
assert_eq!(value["userns_grant"], grant.wire_name());
assert!(report
.render()
.contains(&format!("userns grant: {}", grant.wire_name())));
}
let value: serde_json::Value = serde_json::to_value(&baseline).unwrap();
assert!(value["userns_grant"].is_null());
assert!(baseline.render().contains("userns grant: n/a"));
}
#[test]
fn rendered_report_names_both_classes_and_the_refusal() {
let rendered = report_for(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::KernelFull,
Some(&blocked(SealedBlocker::AppArmorProfileMissing)),
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Sealed,
)),
IoMaxReport::default(),
)
.render();
assert!(rendered.contains("declared: sealed"));
assert!(rendered.contains("achieved: scoped"));
assert!(rendered.contains("floor met: no"));
assert!(rendered.contains("mur-sealed"));
assert!(rendered.contains("would refuse to launch"));
}
#[test]
fn an_export_never_changes_what_the_report_says_about_containment() {
let export = FileExport {
root: "out/".to_string(),
mode: ExportMode::ReadOnly,
max_bytes: 10 * 1024 * 1024,
};
for tier in ALL_TIERS {
for workdir_exec in [false, true] {
let policy = CapabilityPolicy {
workdir_exec_allowed: workdir_exec,
..sample_policy()
};
let without = report_for(
&policy,
ContainmentClass::Scoped,
*tier,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
*tier,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
let with = report_for(
&policy,
ContainmentClass::Scoped,
*tier,
None,
None,
None,
Some(&Exports {
files: Some(export.clone()),
peer_files: None,
peer_tasks: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
*tier,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
assert_eq!(
with.achieved_containment, without.achieved_containment,
"tier {tier:?}, workdir_exec {workdir_exec}"
);
assert_eq!(with.enforcement_tier, without.enforcement_tier);
assert_eq!(with.floor_met, without.floor_met);
assert_eq!(with.shortfall_reason, without.shortfall_reason);
assert_eq!(
ScopeReport {
exports_files: None,
..with.clone()
},
without
);
assert_eq!(
with.exports_files,
Some(ExportsFilesReport {
root: "out/".to_string(),
mode: ExportMode::ReadOnly,
max_bytes: 10 * 1024 * 1024,
})
);
}
}
}
#[test]
fn peer_handoff_never_changes_what_the_report_says_about_containment() {
let peer_files = PeerFilesExport {
root: "out/".to_string(),
max_ttl_secs: Some(900),
max_bytes: 10 * 1024 * 1024,
};
for tier in ALL_TIERS {
for workdir_exec in [false, true] {
let without_policy = CapabilityPolicy {
workdir_exec_allowed: workdir_exec,
..sample_policy()
};
let with_policy = CapabilityPolicy {
peer_fetch_allow: vec!["localhost:41234".to_string()],
..without_policy.clone()
};
let without = report_for(
&without_policy,
ContainmentClass::Scoped,
*tier,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
*tier,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
let with = report_for(
&with_policy,
ContainmentClass::Scoped,
*tier,
None,
None,
None,
Some(&Exports {
files: None,
peer_files: Some(peer_files.clone()),
peer_tasks: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
*tier,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
assert_eq!(
with.achieved_containment, without.achieved_containment,
"tier {tier:?}, workdir_exec {workdir_exec}"
);
assert_eq!(with.enforcement_tier, without.enforcement_tier);
assert_eq!(with.floor_met, without.floor_met);
assert_eq!(with.shortfall_reason, without.shortfall_reason);
assert_eq!(
ScopeReport {
peer_files: None,
peer_fetch_allow: Vec::new(),
..with.clone()
},
without
);
assert_eq!(
with.peer_files,
Some(PeerFilesReport {
root: "out/".to_string(),
max_ttl_secs: 900,
max_bytes: 10 * 1024 * 1024,
})
);
assert_eq!(with.peer_fetch_allow, vec!["localhost:41234".to_string()]);
assert_eq!(with.network_allow, without.network_allow);
}
}
}
#[test]
fn the_peer_fields_are_always_serialized() {
let undeclared: serde_json::Value = serde_json::to_value(report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSealed,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
))
.unwrap();
assert_eq!(undeclared["peer_files"], serde_json::Value::Null);
assert_eq!(undeclared["peer_fetch_allow"], serde_json::json!([]));
let declared: serde_json::Value = serde_json::to_value(report_for(
&CapabilityPolicy {
peer_fetch_allow: vec!["localhost:41234".to_string()],
..sample_policy()
},
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
Some(&Exports {
files: None,
peer_files: Some(PeerFilesExport {
root: "out/".to_string(),
max_ttl_secs: None,
max_bytes: 10 * 1024 * 1024,
}),
peer_tasks: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSealed,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
))
.unwrap();
assert_eq!(
declared["peer_files"],
serde_json::json!({"root": "out/", "max_ttl_secs": 3600, "max_bytes": 10_485_760})
);
assert_eq!(
declared["peer_fetch_allow"],
serde_json::json!(["localhost:41234"])
);
}
#[test]
fn render_names_peer_handoff_in_both_directions() {
let undeclared = report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSealed,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
)
.render();
assert!(undeclared.contains("Peer handoff"));
assert!(undeclared.contains("exports.peer_files: <none>"));
assert!(undeclared.contains("peer_fetch allow: <none>"));
let declared = report_for(
&CapabilityPolicy {
peer_fetch_allow: vec!["localhost:41234".to_string()],
..sample_policy()
},
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
Some(&Exports {
files: None,
peer_files: Some(PeerFilesExport {
root: "out/".to_string(),
max_ttl_secs: Some(900),
max_bytes: 4096,
}),
peer_tasks: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSealed,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
)
.render();
assert!(declared.contains("exports.peer_files root: out/"));
assert!(declared.contains("max handle ttl: 900s"));
assert!(declared.contains("max bytes: 4096 (per file)"));
assert!(declared.contains("- localhost:41234"));
}
fn report_with_peer_tasks(tier: EnforcementTier, peer_tasks: Option<bool>) -> ScopeReport {
report_for(
&sample_policy(),
ContainmentClass::Scoped,
tier,
None,
None,
None,
peer_tasks
.map(|accept| Exports {
files: None,
peer_files: None,
peer_tasks: Some(murmur_artifact::PeerTasksExport { accept }),
})
.as_ref(),
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
tier,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
)
}
#[test]
fn peer_tasks_is_reported_and_moves_no_other_field() {
for tier in ALL_TIERS {
let absent = report_with_peer_tasks(*tier, None);
let refused = report_with_peer_tasks(*tier, Some(false));
let accepted = report_with_peer_tasks(*tier, Some(true));
assert!(!absent.peer_tasks);
assert_eq!(refused, absent, "tier {tier:?}");
assert!(accepted.peer_tasks);
assert_eq!(accepted.achieved_containment, absent.achieved_containment);
assert_eq!(accepted.enforcement_tier, absent.enforcement_tier);
assert_eq!(accepted.floor_met, absent.floor_met);
assert_eq!(
ScopeReport {
peer_tasks: false,
..accepted.clone()
},
absent
);
}
}
#[test]
fn peer_tasks_is_always_serialized() {
let tier = EnforcementTier::KernelFull;
let absent = serde_json::to_value(report_with_peer_tasks(tier, None)).unwrap();
assert_eq!(absent["peer_tasks"], serde_json::json!(false));
let accepted = serde_json::to_value(report_with_peer_tasks(tier, Some(true))).unwrap();
assert_eq!(accepted["peer_tasks"], serde_json::json!(true));
}
#[test]
fn render_names_the_peer_task_posture_under_peer_handoff() {
let tier = EnforcementTier::KernelFull;
for (declared, line) in [
(None, " exports.peer_tasks: refuse\n"),
(Some(false), " exports.peer_tasks: refuse\n"),
(Some(true), " exports.peer_tasks: accept\n"),
] {
let rendered = report_with_peer_tasks(tier, declared).render();
let section = rendered
.split("\nPeer handoff\n")
.nth(1)
.and_then(|rest| rest.split("\n\n").next())
.unwrap_or_else(|| panic!("no Peer handoff section:\n{rendered}"));
assert!(
section.contains(line.trim_start_matches(' ').trim_end()),
"{declared:?}:\n{section}"
);
assert!(rendered.contains(line), "{declared:?}:\n{rendered}");
}
}
#[test]
fn a_state_store_is_reported_and_moves_no_other_field() {
let policy = sample_policy();
let stores = vec![StateStoreReport {
artifact: "murmur/tool-corpus".to_string(),
store: "shey".to_string(),
host_path: "/home/dev/.murmur/state/shey".to_string(),
}];
let without = report_for(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
let with = report_for(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
stores.clone(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
assert_eq!(with.state_stores, stores);
assert!(without.state_stores.is_empty());
assert_eq!(with.achieved_containment, without.achieved_containment);
assert_eq!(with.floor_met, without.floor_met);
assert_eq!(with.enforcement_tier, without.enforcement_tier);
assert_eq!(with.shortfall_reason, without.shortfall_reason);
assert_eq!(
ScopeReport {
state_stores: Vec::new(),
..with.clone()
},
without
);
assert!(with
.render()
.contains("- murmur/tool-corpus: shey -> /home/dev/.murmur/state/shey"));
assert!(without.render().contains("state stores: <none>"));
let json = serde_json::to_value(&without).unwrap();
assert_eq!(json["state_stores"], serde_json::json!([]));
assert_eq!(
serde_json::to_value(&with).unwrap()["state_stores"],
serde_json::json!([{
"artifact": "murmur/tool-corpus",
"store": "shey",
"host_path": "/home/dev/.murmur/state/shey",
}])
);
}
fn extra_roots_preopen(artifact: &str, role: &str, scope: Option<&str>) -> PreopenReport {
PreopenReport {
artifact: artifact.to_string(),
role: role.to_string(),
scope: scope.map(str::to_string),
surface: if scope.is_some() {
PreopenSurface::ScopedSubtree
} else {
PreopenSurface::WholeWorkdir
},
runtime_dir_reach: (role == "tool" && scope.is_none())
.then_some(RuntimeDirReach::MurmurRuntime),
extra_roots: Vec::new(),
}
}
fn two_host_folders() -> Vec<crate::guest_roots::GuestRoot> {
let host_path = |path: &str, access| crate::host_paths::HostPath {
declared: path.to_string(),
path: std::path::PathBuf::from(path),
access,
credential_locations: Vec::new(),
};
crate::guest_roots::extra_guest_roots_for(
ContainmentClass::Advisory,
&[
host_path("/data/ref", crate::host_paths::HostPathAccess::Read),
host_path("/data/out", crate::host_paths::HostPathAccess::ReadWrite),
],
)
}
#[test]
fn extra_roots_serialize_with_four_keys_always_present() {
let mut reports = vec![
extra_roots_preopen("probe", "tool", None),
extra_roots_preopen("cache-tool", "tool", Some("cache")),
extra_roots_preopen("driver", "driver", None),
];
attach_extra_roots(&mut reports, &two_host_folders());
let json = serde_json::to_value(&reports).unwrap();
assert_eq!(
json[0]["extra_roots"],
serde_json::json!([
{
"guest_path": "murmur-host/data/ref",
"path": "/data/ref",
"access": "read",
"source": "host_paths",
},
{
"guest_path": "murmur-host/data/out",
"path": "/data/out",
"access": "read-write",
"source": "host_paths",
},
])
);
assert_eq!(json[1]["extra_roots"], serde_json::json!([]));
assert_eq!(json[2]["extra_roots"], serde_json::json!([]));
let mut reports = vec![extra_roots_preopen("probe", "tool", None)];
attach_extra_roots(
&mut reports,
&crate::guest_roots::extra_guest_roots_for(ContainmentClass::Unconfined, &[]),
);
assert_eq!(
serde_json::to_value(&reports[0].extra_roots).unwrap(),
serde_json::json!([{
"guest_path": "murmur-host",
"path": "/",
"access": "read-write",
"source": "unconfined",
}])
);
}
#[test]
fn extra_roots_render_after_the_runtime_directory() {
let mut reports = vec![extra_roots_preopen("wasm-tool", "tool", None)];
attach_extra_roots(&mut reports, &two_host_folders());
assert_eq!(reports[0].extra_roots.len(), 2);
assert!(
reports[0].render().ends_with(
"; runtime directory read-only at murmur-runtime/; host folders under \
murmur-host/: /data/ref (read), /data/out (read-write)"
),
"{}",
reports[0].render()
);
let mut whole = vec![extra_roots_preopen("wasm-tool", "tool", None)];
attach_extra_roots(
&mut whole,
&crate::guest_roots::extra_guest_roots_for(ContainmentClass::Unconfined, &[]),
);
assert!(whole[0].render().ends_with(
"; runtime directory read-only at murmur-runtime/; whole host read-write \
under murmur-host/"
));
}
#[test]
fn extra_roots_absent_leave_the_line_unchanged() {
let mut reports = vec![extra_roots_preopen("wasm-tool", "tool", None)];
attach_extra_roots(&mut reports, &[]);
assert_eq!(
reports[0].render(),
"wasm-tool (tool): the whole accessible workdir — no capabilities.filesystem.scope \
declared; runtime directory read-only at murmur-runtime/"
);
}
#[test]
fn preopens_are_reported_and_move_no_other_field() {
let policy = sample_policy();
let preopens = vec![
PreopenReport {
artifact: "notes-tool".to_string(),
role: "tool".to_string(),
scope: None,
surface: PreopenSurface::WholeWorkdir,
runtime_dir_reach: Some(RuntimeDirReach::MurmurRuntime),
extra_roots: Vec::new(),
},
PreopenReport {
artifact: "cache-tool".to_string(),
role: "tool".to_string(),
scope: Some("cache".to_string()),
surface: PreopenSurface::ScopedSubtree,
runtime_dir_reach: None,
extra_roots: Vec::new(),
},
PreopenReport {
artifact: "telemetry-hook".to_string(),
role: "hook".to_string(),
scope: None,
surface: PreopenSurface::Nothing,
runtime_dir_reach: None,
extra_roots: Vec::new(),
},
];
let without = report_for(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
let with = report_for(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
preopens.clone(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
assert_eq!(with.preopens, preopens);
assert!(without.preopens.is_empty());
assert_eq!(
ScopeReport {
preopens: Vec::new(),
routes: without.routes.clone(),
..with.clone()
},
without
);
let rendered = with.render();
assert!(rendered.contains(
"- notes-tool (tool): the whole accessible workdir — no capabilities.filesystem.scope \
declared; runtime directory read-only at murmur-runtime/"
));
assert!(rendered.contains(
"- cache-tool (tool): one subtree of the accessible workdir — \
capabilities.filesystem.scope: cache"
));
assert!(rendered.contains(
"- telemetry-hook (hook): nothing preopened — no capabilities.filesystem.scope declared"
));
assert!(without.render().contains("preopens: <none>"));
assert_eq!(
serde_json::to_value(&without).unwrap()["preopens"],
serde_json::json!([])
);
assert_eq!(
serde_json::to_value(&with).unwrap()["preopens"],
serde_json::json!([
{
"artifact": "notes-tool",
"role": "tool",
"scope": null,
"surface": "whole-workdir",
"runtime_dir_reach": "murmur-runtime",
"extra_roots": [],
},
{
"artifact": "cache-tool",
"role": "tool",
"scope": "cache",
"surface": "scoped-subtree",
"runtime_dir_reach": null,
"extra_roots": [],
},
{
"artifact": "telemetry-hook",
"role": "hook",
"scope": null,
"surface": "nothing",
"runtime_dir_reach": null,
"extra_roots": [],
},
])
);
}
#[test]
fn exports_files_is_always_serialized() {
let undeclared = serde_json::to_value(report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
))
.unwrap();
assert_eq!(undeclared["exports_files"], serde_json::Value::Null);
let declared = serde_json::to_value(report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
None,
Some(&Exports {
files: Some(FileExport {
root: "out/".to_string(),
mode: ExportMode::ReadOnly,
max_bytes: 10_485_760,
}),
peer_files: None,
peer_tasks: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
))
.unwrap();
assert_eq!(
declared["exports_files"],
serde_json::json!({"root": "out/", "mode": "read-only", "max_bytes": 10_485_760})
);
}
#[test]
fn render_names_the_resource_plane_in_both_directions() {
let undeclared = report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
)
.render();
assert!(
undeclared.contains("Resource plane") && undeclared.contains("exports.files: <none>"),
"{undeclared}"
);
let declared = report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
None,
Some(&Exports {
files: Some(FileExport {
root: "out/".to_string(),
mode: ExportMode::ReadOnly,
max_bytes: 10_485_760,
}),
peer_files: None,
peer_tasks: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
)
.render();
assert!(declared.contains("exports.files root: out/"), "{declared}");
assert!(
declared.contains("mode: read-only"),
"{declared}"
);
assert!(
declared.contains("max bytes: 10485760 (per file)"),
"{declared}"
);
}
fn read_only_report(read_only: &[&str], shell_allow: &[&str]) -> ScopeReport {
let policy = CapabilityPolicy {
read_only_paths: read_only.iter().map(|p| (*p).to_string()).collect(),
shell_allow: shell_allow.iter().map(|b| (*b).to_string()).collect(),
..CapabilityPolicy::default()
};
report_for(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
)
}
#[test]
fn declared_read_only_with_no_interpreter_reports_as_enforced() {
let report = read_only_report(&["tests", "bench/fixtures"], &["git"]);
assert_eq!(
report.read_only_paths,
vec!["tests".to_string(), "bench/fixtures".to_string()]
);
assert!(report.read_only_advisory_for.is_empty());
let rendered = report.render();
assert!(rendered.contains(" read_only:\n - tests\n - bench/fixtures\n"));
assert!(rendered.contains(
"read_only enforcement: enforced for every tool call and every shell command the \
dispatch check can read"
));
assert!(!rendered.contains("advisory against"));
}
#[test]
fn an_allowlisted_interpreter_is_named_as_advisory() {
let report = read_only_report(&["tests"], &["git", "python3"]);
assert_eq!(report.read_only_advisory_for, vec!["python3".to_string()]);
let rendered = report.render();
assert!(rendered.contains("read_only enforcement: advisory against python3"));
assert!(rendered
.contains("enforced for every tool call and for every shell command it can read"));
}
#[test]
fn the_reported_advisory_set_is_the_set_the_warning_fires_for() {
let shell_allow: Vec<String> = ["git", "python3", "node", "make", "bash"]
.iter()
.map(|b| (*b).to_string())
.collect();
let declared = vec!["tests".to_string()];
assert_eq!(
read_only_report(&["tests"], &["git", "python3", "node", "make", "bash"])
.read_only_advisory_for,
read_only_advisory_for(&declared, &shell_allow)
);
assert!(read_only_advisory_for(&[], &shell_allow).is_empty());
}
#[test]
fn both_read_only_keys_are_always_serialized_arrays() {
let json = serde_json::to_value(read_only_report(&[], &["python3"])).unwrap();
assert_eq!(json["read_only_paths"], serde_json::json!([]));
assert_eq!(json["read_only_advisory_for"], serde_json::json!([]));
let json = serde_json::to_value(read_only_report(&["tests"], &["python3"])).unwrap();
assert_eq!(json["read_only_paths"], serde_json::json!(["tests"]));
assert_eq!(
json["read_only_advisory_for"],
serde_json::json!(["python3"])
);
}
#[test]
fn an_undeclared_read_only_renders_none_and_no_enforcement_line() {
let rendered = read_only_report(&[], &["python3"]).render();
assert!(rendered.contains(" read_only: <none>\n"));
assert!(!rendered.contains("read_only enforcement:"));
}
#[test]
fn every_capability_policy_field_is_reported_or_exempted() {
const EXEMPT: &[(&str, &str)] = &[
(
"shell_strip_env",
"Names host variables removed from the shell child's environment on top of the \
credential patterns the runtime always strips. It subtracts from what the \
capsule sees and can open nothing, so reporting it beside the grants would pad \
the list an operator audits for reach.",
),
(
"shell_baseline_env",
"The fixed variables the shell child starts with (PATH, HOME, …). It shapes the \
environment of a subprocess the capsule was already granted, and names no host \
resource `shell allow` and `env allow` do not already account for.",
),
(
"limits",
"Execution ceilings for a WASM guest — memory, table elements, instance count, \
deadline. A bound on how much the capsule may consume, never a reach it gains.",
),
(
"resources",
"The same, for subprocesses: rlimits, the Linux cgroup v2 scope and the \
workdir size ceiling. A ceiling, not a grant, on exactly `limits`' terms.",
),
(
"declared_deadline_seconds",
"The undefaulted form of `limits.deadline_seconds`, retained only so hook calls \
can apply a lower default. Reporting it would report one ceiling twice.",
),
(
"state_declared",
"Records that a capsule-wide `capabilities.state` block was declared where \
nothing reads it, which `W-SEC-014` says in words. The stores that actually \
exist are reported per artifact in `state_stores`.",
),
(
"conversation_declared",
"Records a capsule-wide `capabilities.conversation` block that reaches no \
artifact, on `state_declared`'s terms; `W-SEC-016` is what an operator needs, \
and the grant itself is per artifact.",
),
];
const REPORTED: &[(&str, &str)] = &[
("network_allow", "network_allow"),
("peer_fetch_allow", "peer_fetch_allow"),
("unix_sockets_allowed", "unix_sockets"),
("filesystem_scope", "filesystem_scope"),
("workdir_exec_allowed", "workdir_exec"),
("read_only_paths", "read_only_paths"),
("shell_allow", "shell_allow"),
("spawn_allow", "spawn_allow"),
("spawn_workdir_snapshot", "spawn_workdir_snapshot"),
("shell_interpreter_runtime", "interpreter_runtime_grants"),
(
"shell_derived_interpreter_runtime",
"derived_interpreter_runtime",
),
("shell_trust_store", "trust_store"),
("host_paths", "host_paths"),
("shell_staged_runtime", "staged_runtime_grants"),
("env_allow", "env_allow"),
("plan_submit", "plan_submit"),
("install_skill", "install_skill"),
("install_tool", "install_tool"),
("install_from_sources", "install_from"),
("containment_floor", "declared_containment"),
("declared_resources", "unconfined"),
];
let declared = capability_policy_field_names(include_str!("types.rs"));
assert!(
declared.len() > 10,
"failed to read CapabilityPolicy's fields out of types.rs (found {declared:?})"
);
for field in &declared {
let reported = REPORTED.iter().any(|(policy, _)| policy == field);
let exempt = EXEMPT.iter().any(|(policy, _)| policy == field);
assert!(
reported || exempt,
"CapabilityPolicy::{field} is a grant no --explain-scope report mentions. Either \
add it to ScopeReport and to this test's REPORTED table, or add it to EXEMPT \
with a written reason for why an operator does not need to see it."
);
assert!(
!(reported && exempt),
"CapabilityPolicy::{field} is listed as both reported and exempt"
);
}
for (policy_field, _) in REPORTED.iter().chain(EXEMPT.iter()) {
assert!(
declared.iter().any(|field| field == policy_field),
"the table names CapabilityPolicy::{policy_field}, which types.rs no longer \
declares — remove the stale row"
);
}
let json = serde_json::to_value(report_for(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
))
.unwrap();
let object = json.as_object().expect("a scope report is a JSON object");
for (policy_field, report_key) in REPORTED {
assert!(
object.contains_key(*report_key),
"the table maps CapabilityPolicy::{policy_field} to a report key \
'{report_key}' that a serialized ScopeReport does not carry"
);
}
}
fn boundary_report(declared: ContainmentClass, tier: EnforcementTier) -> ScopeReport {
report_for(
&sample_policy(),
declared,
tier,
None,
None,
Some(6),
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(tier, declared)),
IoMaxReport::default(),
)
}
fn sample_snapshot() -> crate::workdir_snapshot::WorkdirSnapshot {
crate::workdir_snapshot::WorkdirSnapshot {
source: "/home/you/project".to_string(),
files: 12,
dirs: 3,
symlinks: 1,
repointed_symlinks: 0,
bytes: 40960,
skipped_count: 1,
skipped: vec![crate::workdir_snapshot::SkippedEntry {
path: "run/pipe".to_string(),
reason: crate::workdir_snapshot::SkipReason::Fifo,
}],
}
}
#[test]
fn spawn_workdir_snapshot_and_parent_workdir_are_always_reported() {
let report = boundary_report(ContainmentClass::Scoped, EnforcementTier::KernelFull);
let json = serde_json::to_value(&report).unwrap();
assert_eq!(json["spawn_workdir_snapshot"], serde_json::json!([]));
assert!(json.as_object().unwrap().contains_key("parent_workdir"));
assert_eq!(json["parent_workdir"], serde_json::Value::Null);
let rendered = report.render();
let snapshot_line = rendered
.find("\n spawn workdir snapshot: <none>\n")
.unwrap_or_else(|| panic!("{rendered}"));
assert!(
rendered.find("\n spawn allow:").unwrap() < snapshot_line,
"{rendered}"
);
assert!(!rendered.contains("Parent work folder"), "{rendered}");
let mut policy = sample_policy();
policy.spawn_allow = vec!["local/reviewer".to_string(), "local/worker".to_string()];
policy.spawn_workdir_snapshot = vec!["local/reviewer".to_string()];
let report = report_for(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
Some(6),
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
)),
IoMaxReport::default(),
);
assert_eq!(report.spawn_workdir_snapshot, vec!["local/reviewer"]);
assert_eq!(
serde_json::to_value(&report).unwrap()["spawn_workdir_snapshot"],
serde_json::json!(["local/reviewer"])
);
assert!(
report
.render()
.contains(" spawn workdir snapshot:\n - local/reviewer\n"),
"{}",
report.render()
);
}
#[test]
fn parent_workdir_restriction_follows_the_applied_tier() {
for (declared, tier, restriction) in [
(
ContainmentClass::Sealed,
EnforcementTier::KernelSealed,
"absent",
),
(
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
"enforced",
),
(
ContainmentClass::Advisory,
EnforcementTier::KernelSeccompOnly,
"advisory",
),
(
ContainmentClass::Advisory,
EnforcementTier::EnvironmentOnly,
"advisory",
),
] {
let report =
boundary_report(declared, tier).with_parent_workdir(Some(sample_snapshot()));
let json = serde_json::to_value(&report).unwrap();
let parent = &json["parent_workdir"];
assert_eq!(parent["restriction"], restriction, "{tier:?}");
assert_eq!(
parent["restriction"],
json["filesystem_boundary"]["restriction"]
);
assert_eq!(parent["source"], "/home/you/project");
assert_eq!(parent["files"], 12);
assert_eq!(parent["skipped"][0]["reason"], "fifo");
let rendered = report.render();
assert!(
rendered.contains(
"\nParent work folder\n source: /home/you/project\n copied: \
12 files, 3 directories, 1 symlinks (0 re-pointed), 40960 bytes\n \
skipped: 1\n - run/pipe (fifo)\n restriction: "
),
"{rendered}"
);
assert!(
rendered.contains(&format!(" restriction: {restriction}: ")),
"{rendered}"
);
}
assert_eq!(
boundary_report(ContainmentClass::Scoped, EnforcementTier::KernelFull)
.with_parent_workdir(None)
.parent_workdir,
None
);
}
fn presence_cases() -> Vec<(&'static str, RouteInputs, Vec<Route>)> {
vec![
("nothing", RouteInputs::default(), vec![]),
(
"one tool",
RouteInputs {
tools: vec!["t".to_string()],
..RouteInputs::default()
},
vec![Route::WasmTools],
),
(
"a driver",
RouteInputs {
drivers: vec!["d".to_string()],
..RouteInputs::default()
},
vec![Route::Drivers],
),
(
"a hook",
RouteInputs {
hooks: vec!["h".to_string()],
..RouteInputs::default()
},
vec![Route::Hooks],
),
(
"install.tool",
RouteInputs {
install_tool: true,
..RouteInputs::default()
},
vec![Route::WasmTools],
),
(
"transport: process",
RouteInputs {
process_transport: true,
..RouteInputs::default()
},
vec![Route::ProcessHarness, Route::HarnessTools],
),
(
"plan.submit without shell.allow",
RouteInputs {
plan_submit: true,
..RouteInputs::default()
},
vec![],
),
(
"plan.submit with shell.allow",
RouteInputs {
plan_submit: true,
shell_allow: true,
..RouteInputs::default()
},
vec![Route::Shell, Route::PlanShellSteps],
),
(
"shell.allow alone",
RouteInputs {
shell_allow: true,
..RouteInputs::default()
},
vec![Route::Shell],
),
]
}
#[test]
fn the_route_table_holds_its_invariants_on_every_input() {
for tier in ALL_TIERS {
for declared in [
ContainmentClass::Advisory,
ContainmentClass::Scoped,
ContainmentClass::Sealed,
] {
let mechanism =
ContainmentDecision::decide(*tier, declared, false, false).enforcement_tier();
for (case, inputs, present) in presence_cases() {
let rows = route_table(&inputs, mechanism);
assert_eq!(
rows.iter().map(|row| row.route).collect::<Vec<_>>(),
Route::ALL.to_vec(),
"{case}"
);
for row in &rows {
let covered = matches!(row.route, Route::Shell | Route::PlanShellSteps);
assert_eq!(row.under_class, covered, "{case}: {row:?}");
if covered {
assert_eq!(row.mechanism, mechanism, "{case}: {row:?}");
assert_eq!(row.note, None, "{case}: {row:?}");
} else {
assert_eq!(row.mechanism, HOST_PROCESS_MECHANISM, "{case}: {row:?}");
assert!(row.note.is_some(), "{case}: {row:?}");
}
assert_eq!(
row.present,
present.contains(&row.route),
"{case} on {tier:?}/{declared}: {row:?}"
);
}
}
}
}
}
#[test]
fn each_uncovered_route_names_what_holds_it() {
let notes = |route: Route| route.note().unwrap();
for route in [Route::WasmTools, Route::Drivers] {
assert!(notes(route).contains("inside the mur process"), "{route:?}");
assert!(notes(route).contains("WASI preopens"), "{route:?}");
assert!(
notes(route).contains("host-side network allowlist only"),
"{route:?}"
);
}
assert!(notes(Route::Hooks).contains("its own capabilities block declares"));
let harness = notes(Route::ProcessHarness);
for word in [
"cleared environment",
"Landlock",
"seccomp",
"network namespace",
"rlimits",
"cgroup with no limits",
] {
assert!(harness.contains(word), "{word}: {harness}");
}
assert!(notes(Route::HarnessTools).contains("the user's authority"));
assert!(notes(Route::HarnessTools).contains("arguments the process driver passes"));
}
#[test]
fn route_inputs_read_roles_off_the_preopens() {
let preopen = |artifact: &str, role: &str| PreopenReport {
artifact: artifact.to_string(),
role: role.to_string(),
scope: None,
surface: PreopenSurface::WholeWorkdir,
runtime_dir_reach: None,
extra_roots: Vec::new(),
};
let preopens = vec![
preopen("fs-tool", "tool"),
preopen("browser", "tool"),
preopen("driver-anthropic", "driver"),
preopen("audit", "hook"),
];
let inputs = RouteInputs::new(&[], false, &[], &preopens, Some("http"));
let rows = route_table(&inputs, "landlock+seccomp");
let row = |route: Route| rows.iter().find(|row| row.route == route).unwrap();
assert_eq!(row(Route::WasmTools).artifacts, vec!["fs-tool", "browser"]);
assert_eq!(row(Route::Drivers).artifacts, vec!["driver-anthropic"]);
assert_eq!(row(Route::Hooks).artifacts, vec!["audit"]);
assert!(!row(Route::ProcessHarness).present);
assert!(RouteInputs::new(&[], false, &[], &[], Some("process")).process_transport);
}
#[test]
fn the_routes_json_is_pinned() {
let report = boundary_report(ContainmentClass::Advisory, EnforcementTier::KernelFull)
.with_routes(Some("process"));
let json = serde_json::to_value(&report).unwrap();
assert!(json.get("mechanism_covers").is_none(), "{json}");
let routes = json["routes"].as_array().expect("routes is an array");
assert_eq!(routes.len(), 7);
let shell_present = !report.shell_allow.is_empty();
assert_eq!(
routes[0],
serde_json::json!({
"route": "shell",
"present": shell_present,
"under_class": true,
"mechanism": "landlock+seccomp",
"note": null,
"artifacts": [],
})
);
assert_eq!(
routes[5],
serde_json::json!({
"route": "process-harness",
"present": true,
"under_class": false,
"mechanism": "host-process",
"note": Route::ProcessHarness.note(),
"artifacts": [],
})
);
assert_eq!(
routes
.iter()
.map(|row| row["route"].as_str().unwrap())
.collect::<Vec<_>>(),
[
"shell",
"plan-shell-steps",
"wasm-tools",
"drivers",
"hooks",
"process-harness",
"harness-tools"
]
);
assert_eq!(
json["cgroup_scope"],
serde_json::json!({"status": "not-probed"})
);
}
#[test]
fn the_routes_block_follows_the_containment_block() {
let preopens = vec![PreopenReport {
artifact: "fs-tool".to_string(),
role: "tool".to_string(),
scope: None,
surface: PreopenSurface::WholeWorkdir,
runtime_dir_reach: None,
extra_roots: Vec::new(),
}];
let report = report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
Some(6),
None,
Vec::new(),
Vec::new(),
preopens,
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(EnforcementTier::KernelFull),
IoMaxReport::default(),
)
.with_routes(None);
let rendered = report.render();
assert!(!rendered.contains("covers:"), "{rendered}");
let routes = rendered.find("\nRoutes\n").expect("a Routes block");
assert!(
routes > rendered.find(" mechanism: ").unwrap(),
"{rendered}"
);
assert!(
routes < rendered.find("\nRuntime directory\n").unwrap(),
"{rendered}"
);
assert!(
rendered.contains(" shell commands present landlock+seccomp\n"),
"{rendered}"
);
assert!(
rendered.contains(
" WASM tools present host process, not confined by the class (fs-tool)\n \
Runs inside the mur process"
),
"{rendered}"
);
assert!(
rendered
.contains(" process harness absent host process, not confined by the class\n"),
"{rendered}"
);
assert!(
rendered.contains(" cgroup scope: not-probed\n"),
"{rendered}"
);
}
#[test]
fn the_uncovered_route_warning_names_exactly_the_present_uncovered_routes() {
let inputs = RouteInputs {
shell_allow: true,
tools: vec!["fs-tool".to_string()],
drivers: vec!["driver-anthropic".to_string()],
..RouteInputs::default()
};
let routes = route_table(&inputs, "landlock+seccomp");
assert_eq!(
uncovered_route_warning(ContainmentClass::Advisory, &routes),
None
);
let covered_only = route_table(
&RouteInputs {
shell_allow: true,
plan_submit: true,
..RouteInputs::default()
},
"landlock+seccomp",
);
assert_eq!(
uncovered_route_warning(ContainmentClass::Scoped, &covered_only),
None
);
for declared in [ContainmentClass::Scoped, ContainmentClass::Sealed] {
let message = uncovered_route_warning(declared, &routes).expect("fires");
assert!(
message.contains(&format!("the declared {declared} floor")),
"{message}"
);
assert!(
message.contains(
"WASM tools (fs-tool) and drivers (driver-anthropic) run outside the class"
),
"{message}"
);
assert!(message.contains("nothing is refused"), "{message}");
for absent in ["hooks", "process harness", "harness tools"] {
assert!(!message.contains(absent), "{absent}: {message}");
}
}
let process = route_table(
&RouteInputs {
process_transport: true,
drivers: vec!["driver-claude-code".to_string()],
..RouteInputs::default()
},
"landlock+seccomp",
);
let message = uncovered_route_warning(ContainmentClass::Scoped, &process).unwrap();
assert!(
message.contains(
"drivers (driver-claude-code), the process harness and harness tools run"
),
"{message}"
);
}
#[test]
fn the_uncovered_route_warning_reaches_the_bootstrap_log() {
let temp = tempfile::tempdir().unwrap();
let routes = route_table(
&RouteInputs {
drivers: vec!["d".to_string()],
..RouteInputs::default()
},
"landlock+seccomp",
);
warn_on_uncovered_routes(
ContainmentClass::Sealed,
&routes,
Some(&crate::workdir_fs::WorkdirRoot::open(temp.path()).unwrap()),
);
let log = std::fs::read_to_string(temp.path().join("logs").join("bootstrap.log"))
.unwrap_or_default();
assert_eq!(log.matches("warning[W-SEC-034]").count(), 1, "{log}");
assert!(log.contains("drivers (d)"), "{log}");
assert!(log.contains("#w-sec-034"), "{log}");
}
#[test]
fn an_advisory_session_says_home_is_relocated_not_protected() {
for tier in [
EnforcementTier::EnvironmentOnly,
EnforcementTier::KernelSeccompOnly,
] {
let boundary = filesystem_boundary(tier);
let home = boundary
.not_protected
.iter()
.find(|statement| statement.contains("relocated, not protected"))
.unwrap_or_else(|| panic!("{tier:?}: {boundary:?}"));
assert!(home.contains("session's home folder"), "{home}");
assert!(
home.contains("real home directory stays readable and writable by absolute path"),
"{home}"
);
assert!(home.contains("no kernel mechanism"), "{home}");
}
assert!(!filesystem_boundary(EnforcementTier::KernelFull)
.not_protected
.iter()
.any(|statement| statement.contains("relocated, not protected")));
assert!(filesystem_boundary(EnforcementTier::KernelSealed)
.not_protected
.is_empty());
}
#[test]
fn a_session_with_no_kernel_mechanism_reports_what_it_does_not_protect() {
let report = boundary_report(ContainmentClass::Advisory, EnforcementTier::EnvironmentOnly);
assert_eq!(report.achieved_containment, ContainmentClass::Advisory);
assert_eq!(
report.filesystem_boundary.restriction,
FilesystemRestriction::Advisory
);
assert_eq!(report.filesystem_boundary.not_protected.len(), 5);
assert_eq!(
report.filesystem_boundary.not_protected[2],
ADVISORY_HOME_NOT_PROTECTED
);
assert_eq!(
Some(&report.filesystem_boundary.not_protected[3]),
report.runtime_dir.not_protected_because.as_ref()
);
assert_eq!(
Some(&report.filesystem_boundary.not_protected[4]),
report.records_dir.not_protected_because.as_ref()
);
assert!(report.filesystem_boundary.not_protected[4].contains("the records directory"));
let filesystem = &report.filesystem_boundary.not_protected[0];
assert!(filesystem.contains("advisory"), "{filesystem}");
assert!(
filesystem.contains("denies a read or a write outside the work folder"),
"{filesystem}"
);
let rewriting = &report.filesystem_boundary.not_protected[1];
assert!(rewriting.contains("not a boundary"), "{rewriting}");
assert!(rewriting.contains("HOME"), "{rewriting}");
let rendered = report.render();
let heading = rendered
.find("\nNot protected here\n")
.expect("a session that composes no root prints the block");
assert!(
heading > rendered.find("Containment\n").unwrap(),
"the block belongs under Containment:\n{rendered}"
);
assert!(
heading < rendered.find("\nEffective grants\n").unwrap(),
"the block belongs ahead of the grants an operator reads it against:\n{rendered}"
);
for statement in &report.filesystem_boundary.not_protected {
assert!(rendered.contains(statement.as_str()), "{rendered}");
}
}
fn inside_placement() -> RuntimeDirPlacement {
RuntimeDirPlacement {
sessions_dir: std::path::PathBuf::from("/home/dev/.murmur/sessions"),
records_group: std::path::PathBuf::from("/home/dev/.murmur/records/home/dev"),
workdir: std::path::PathBuf::from("/home/dev"),
sessions_inside_workdir: true,
records_inside_workdir: true,
}
}
#[test]
fn runtime_dir_report_makes_one_claim_per_tier() {
let outside = RuntimeDirPlacement::outside_for_test();
for abi in [None, Some(1), Some(2), Some(3), Some(6)] {
let sealed = runtime_dir_report(EnforcementTier::KernelSealed, abi, &outside);
assert!(sealed.read_only_to_commands, "sealed at {abi:?}");
assert_eq!(sealed.enforced_by, Some("mountns+landlock"));
assert_eq!(sealed.not_protected_because, None);
}
let full = runtime_dir_report(EnforcementTier::KernelFull, Some(3), &outside);
assert!(full.read_only_to_commands);
assert_eq!(full.enforced_by, Some("landlock"));
assert_eq!(full.not_protected_because, None);
for abi in [Some(1), Some(2)] {
let old = runtime_dir_report(EnforcementTier::KernelFull, abi, &outside);
assert!(!old.read_only_to_commands);
assert_eq!(old.enforced_by, None);
let reason = old.not_protected_because.unwrap();
assert!(
reason.contains(&format!("Landlock ABI {}", abi.unwrap()))
&& reason.contains("truncate(2)")
&& reason.contains("6.2"),
"{reason}"
);
}
let seccomp = runtime_dir_report(EnforcementTier::KernelSeccompOnly, None, &outside);
assert!(!seccomp.read_only_to_commands);
assert!(seccomp
.not_protected_because
.unwrap()
.contains("no Landlock"));
let none = runtime_dir_report(EnforcementTier::EnvironmentOnly, None, &outside);
assert!(!none.read_only_to_commands);
let reason = none.not_protected_because.unwrap();
assert!(
reason.contains("no kernel containment") && reason.contains("macOS"),
"{reason}"
);
assert_eq!(
full.path, "/home/test/.murmur/sessions/<session-id>/runtime",
"the path carries the literal session-id segment"
);
}
#[test]
fn runtime_dir_report_inside_the_workdir_is_never_protected() {
for tier in ALL_TIERS {
let report = runtime_dir_report(*tier, Some(6), &inside_placement());
assert!(!report.read_only_to_commands, "{tier:?}");
assert_eq!(report.enforced_by, None);
}
let reason =
runtime_dir_report(EnforcementTier::KernelSealed, Some(6), &inside_placement())
.not_protected_because
.unwrap();
assert!(
reason.contains("/home/dev/.murmur/sessions/<session-id>/runtime"),
"{reason}"
);
assert!(reason.contains("inside the workdir /home/dev"), "{reason}");
}
#[test]
fn runtime_dir_inside_the_workdir_caps_the_class_and_refuses_a_kernel_floor() {
let inside = inside_placement();
assert_eq!(
achieved_containment_class(EnforcementTier::KernelSealed, false, true),
ContainmentClass::Advisory
);
assert_eq!(
achieved_containment_class(EnforcementTier::KernelSealed, false, false),
ContainmentClass::Sealed
);
for declared in [ContainmentClass::Scoped, ContainmentClass::Sealed] {
match check_containment_floor(
declared,
ContainmentClass::Advisory,
None,
false,
inside.inside(),
) {
Err(RuntimeError::ContainmentFloorUnmet { reason, .. }) => {
assert!(reason.contains("/home/dev/.murmur/sessions"), "{reason}");
assert!(reason.contains("the workdir /home/dev"), "{reason}");
assert!(
reason.contains("run with a --workdir that does not contain ~/.murmur"),
"{reason}"
);
}
other => panic!("{declared:?}: {other:?}"),
}
}
assert!(check_containment_floor(
ContainmentClass::Advisory,
ContainmentClass::Advisory,
None,
false,
inside.inside()
)
.is_ok());
let report = report_for(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
Some(6),
None,
Vec::new(),
Vec::new(),
Vec::new(),
&inside,
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelFull,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
);
assert_eq!(report.achieved_containment, ContainmentClass::Advisory);
let reason = report.runtime_dir.not_protected_because.clone().unwrap();
assert!(report.filesystem_boundary.not_protected.contains(&reason));
let rendered = report.render();
assert!(
rendered.contains(&format!(" commands: can change it — {reason}\n")),
"{rendered}"
);
}
#[test]
fn runtime_dir_block_renders_after_containment_and_serializes_every_field() {
let report = boundary_report(ContainmentClass::Scoped, EnforcementTier::KernelFull);
let rendered = report.render();
let block =
"Runtime directory\n path: /home/test/.murmur/sessions/<session-id>/runtime\n \
commands: read only (landlock)\n";
let at = rendered.find(block).expect(&rendered);
assert!(at > rendered.find("Containment\n").unwrap());
assert!(at < rendered.find("Effective grants").unwrap());
let sealed = boundary_report(ContainmentClass::Sealed, EnforcementTier::KernelSealed);
assert!(sealed
.render()
.contains(" commands: read only (mountns+landlock)\n"));
let json = serde_json::to_string(&report.runtime_dir).unwrap();
assert_eq!(
json,
r#"{"path":"/home/test/.murmur/sessions/<session-id>/runtime","read_only_to_commands":true,"enforced_by":"landlock","not_protected_because":null}"#
);
}
#[test]
fn a_landlock_session_reports_denial_rather_than_absence() {
let report = boundary_report(ContainmentClass::Advisory, EnforcementTier::KernelFull);
assert_eq!(report.achieved_containment, ContainmentClass::Scoped);
assert_eq!(
report.filesystem_boundary.restriction,
FilesystemRestriction::Enforced
);
assert_eq!(report.filesystem_boundary.not_protected.len(), 2);
let filesystem = &report.filesystem_boundary.not_protected[0];
assert!(
!filesystem.contains("advisory"),
"Landlock mediates the filesystem here: {filesystem}"
);
assert!(
filesystem.contains("denied rather than absent"),
"{filesystem}"
);
for probe in ["stat", "access", "readlink"] {
assert!(filesystem.contains(probe), "{filesystem}");
}
}
#[test]
fn a_sealed_capable_host_still_discloses_landlock_for_a_capsule_that_declared_less() {
let report = boundary_report(ContainmentClass::Advisory, EnforcementTier::KernelSealed);
assert_eq!(report.achieved_containment, ContainmentClass::Scoped);
assert_eq!(report.enforcement_tier, "landlock+seccomp");
assert_eq!(
report.host_containment,
ContainmentClass::Sealed,
"the host can back sealed, which only `host_containment` says"
);
assert_eq!(
report.filesystem_boundary.restriction,
FilesystemRestriction::Enforced,
"no composed root is installed for a capsule that did not declare one"
);
assert_eq!(
report.filesystem_boundary.not_protected,
boundary_report(ContainmentClass::Advisory, EnforcementTier::KernelFull)
.filesystem_boundary
.not_protected,
"the same mechanism gets the same statements on either host"
);
assert!(report.render().contains("Not protected here"));
}
#[test]
fn the_path_rewriting_statement_is_identical_wherever_it_is_carried() {
let advisory =
boundary_report(ContainmentClass::Advisory, EnforcementTier::EnvironmentOnly);
let enforced = boundary_report(ContainmentClass::Advisory, EnforcementTier::KernelFull);
assert_eq!(
advisory.filesystem_boundary.not_protected[1],
enforced.filesystem_boundary.not_protected[1]
);
assert!(
advisory.filesystem_boundary.not_protected[1]
.contains("home/ in its records directory"),
"the statement must name the directory `shell::build_shell_env` really sets HOME to"
);
}
#[test]
fn a_composed_root_carries_the_key_with_nothing_to_disclaim() {
let report = boundary_report(ContainmentClass::Sealed, EnforcementTier::KernelSealed);
assert_eq!(report.achieved_containment, ContainmentClass::Sealed);
assert_eq!(
report.filesystem_boundary.restriction,
FilesystemRestriction::Absent
);
assert!(report.filesystem_boundary.not_protected.is_empty());
assert!(
!report.render().contains("Not protected here"),
"{}",
report.render()
);
let json = serde_json::to_value(&report).unwrap();
assert_eq!(json["filesystem_boundary"]["restriction"], "absent");
assert_eq!(
json["filesystem_boundary"]["not_protected"],
serde_json::json!([]),
"the key is present and empty at absent, never missing"
);
}
#[test]
fn workdir_exec_caps_the_class_without_changing_the_mechanism() {
let policy = CapabilityPolicy {
workdir_exec_allowed: true,
..CapabilityPolicy::default()
};
let report = report_for(
&policy,
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
Some(6),
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
EnforcementTier::KernelSealed,
ContainmentClass::Advisory,
)),
IoMaxReport::default(),
);
assert_eq!(report.enforcement_tier, "landlock+seccomp");
assert_eq!(report.achieved_containment, ContainmentClass::Advisory);
assert_eq!(
report.filesystem_boundary.restriction,
FilesystemRestriction::Enforced
);
assert_eq!(report.filesystem_boundary.not_protected.len(), 2);
assert!(report.render().contains("Not protected here"));
}
#[test]
fn the_restriction_follows_the_tier_the_session_would_install() {
for tier in ALL_TIERS {
for declared in [
ContainmentClass::Advisory,
ContainmentClass::Scoped,
ContainmentClass::Sealed,
] {
let report = boundary_report(declared, *tier);
let expected = match crate::sandbox::applied_tier(*tier, declared) {
EnforcementTier::KernelSealed => FilesystemRestriction::Absent,
EnforcementTier::KernelFull => FilesystemRestriction::Enforced,
EnforcementTier::KernelSeccompOnly | EnforcementTier::EnvironmentOnly => {
FilesystemRestriction::Advisory
}
EnforcementTier::Unconfined => FilesystemRestriction::None,
};
assert_eq!(
report.filesystem_boundary.restriction, expected,
"{tier:?} / {declared:?}"
);
assert_eq!(
report.filesystem_boundary.not_protected.is_empty(),
expected == FilesystemRestriction::Absent,
"{tier:?} / {declared:?}"
);
let wire = match expected {
FilesystemRestriction::Advisory => "advisory",
FilesystemRestriction::Enforced => "enforced",
FilesystemRestriction::Absent => "absent",
FilesystemRestriction::None => "none",
};
assert_eq!(
serde_json::to_value(report.filesystem_boundary.restriction).unwrap(),
serde_json::Value::String(wire.to_string())
);
}
}
}
#[test]
fn the_reported_restriction_matches_what_a_subprocess_actually_reaches() {
if crate::network_namespace::skip_without_egress_namespace(
"the_reported_restriction_matches_what_a_subprocess_actually_reaches",
) {
return;
}
if crate::sandbox::find_on_path("bash").is_none() {
eprintln!(
"[SKIP-HOST] the_reported_restriction_matches_what_a_subprocess_actually_reaches: \
no bash on PATH"
);
return;
}
let outside = tempfile::tempdir().expect("a temp directory outside the workdir");
let secret = outside.path().join("host-file");
std::fs::write(&secret, "PROBE-CONTENTS\n").unwrap();
let secret = secret.to_str().unwrap().to_string();
for declared in [ContainmentClass::Advisory, ContainmentClass::Sealed] {
let workdir = tempfile::tempdir().unwrap();
let policy = CapabilityPolicy {
shell_allow: vec!["bash".to_string()],
..CapabilityPolicy::default()
};
let probe = crate::sandbox::HostProbe::probe();
let host_tier = probe.tier();
let runtime = tempfile::tempdir().unwrap();
let enforcement =
crate::sandbox::ShellEnforcement::resolve_declared(&policy, declared, probe)
.unwrap()
.with_runtime_dir(crate::sandbox::test_runtime_dir(runtime.path()), None);
let report = report_for(
&policy,
declared,
host_tier,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
host_tier, declared,
)),
IoMaxReport::default(),
);
let command = format!(
"if read -r line < '{secret}' 2>/dev/null; then echo \"read=$line\"; \
else echo read=refused; fi; \
if [[ -e '{secret}' ]]; then echo exists=yes; else echo exists=no; fi"
);
let result = crate::shell::execute_shell(
"bash",
&["-c", &command],
&[],
&crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
&policy,
&enforcement,
)
.unwrap_or_else(|error| panic!("declared {declared:?}: {error}"));
let seen = result.stdout;
match report.filesystem_boundary.restriction {
FilesystemRestriction::Advisory => assert!(
seen.contains("read=PROBE-CONTENTS"),
"reported advisory, so nothing denies the read: {seen}"
),
FilesystemRestriction::Enforced => assert!(
seen.contains("read=refused") && seen.contains("exists=yes"),
"reported enforced, so the read is denied and the path still exists: {seen}"
),
FilesystemRestriction::Absent => assert!(
seen.contains("read=refused") && seen.contains("exists=no"),
"reported absent, so the path is not in the composed root: {seen}"
),
FilesystemRestriction::None => assert!(
seen.contains("read=PROBE-CONTENTS"),
"reported none, so nothing denies the read: {seen}"
),
}
}
}
fn capability_policy_field_names(source: &str) -> Vec<String> {
source
.split_once("pub struct CapabilityPolicy {")
.expect("types.rs declares CapabilityPolicy")
.1
.split_once("\n}")
.expect("CapabilityPolicy's declaration is closed at column 0")
.0
.lines()
.filter_map(|line| {
line.strip_prefix(" pub ")
.and_then(|rest| rest.split_once(':'))
.map(|(name, _)| name.to_string())
})
.collect()
}
#[test]
fn the_temp_dir_report_follows_the_per_tier_table() {
use crate::temp_dir::temp_mapping;
let placement = RuntimeDirPlacement::outside_for_test();
let folder = "/home/test/.murmur/records/work/<session-id>/tmp";
let tmp = crate::temp_dir::canonical_tmp();
let elsewhere = std::path::Path::new("/work");
let rows = [
(
EnforcementTier::KernelSealed,
ContainmentClass::Sealed,
true,
elsewhere,
true,
Some("composed-root"),
),
(
EnforcementTier::KernelSealed,
ContainmentClass::Scoped,
true,
elsewhere,
true,
Some("mount-namespace"),
),
(
EnforcementTier::KernelSealed,
ContainmentClass::Advisory,
true,
elsewhere,
true,
Some("mount-namespace"),
),
(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
true,
elsewhere,
true,
Some("mount-namespace"),
),
(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
false,
elsewhere,
false,
None,
),
(
EnforcementTier::KernelSeccompOnly,
ContainmentClass::Advisory,
true,
elsewhere,
true,
Some("mount-namespace"),
),
(
EnforcementTier::KernelSeccompOnly,
ContainmentClass::Advisory,
false,
elsewhere,
false,
None,
),
(
EnforcementTier::EnvironmentOnly,
ContainmentClass::Advisory,
false,
elsewhere,
false,
None,
),
(
EnforcementTier::KernelFull,
ContainmentClass::Scoped,
true,
tmp.as_path(),
false,
None,
),
(
EnforcementTier::KernelSeccompOnly,
ContainmentClass::Advisory,
true,
tmp.as_path(),
false,
None,
),
];
for (tier, declared, usable, workdir, mapped, mapped_by) in rows {
let applied = crate::sandbox::applied_tier(tier, declared);
let mapping = temp_mapping(applied, usable, workdir);
let report = temp_dir_report(mapping, &placement);
let row = format!(
"{tier:?} {declared:?} usable={usable} workdir={}",
workdir.display()
);
assert_eq!(report.path, folder, "{row}");
assert_eq!(report.tmp_mapped, mapped, "{row}");
assert_eq!(report.mapped_by, mapped_by, "{row}");
assert_eq!(report.tmpdir, if mapped { "/tmp" } else { folder }, "{row}");
assert_eq!(
report
.not_mapped_because
.as_deref()
.is_some_and(|reason| !reason.is_empty()),
!mapped,
"{row}"
);
let scope = report_for(
&CapabilityPolicy::default(),
declared,
tier,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&placement,
mapping,
IoMaxReport::default(),
);
assert_eq!(scope.temp_dir, report, "{row}");
}
let reason = |tier, usable, workdir| {
temp_dir_report(temp_mapping(tier, usable, workdir), &placement)
.not_mapped_because
.unwrap()
};
assert!(reason(EnforcementTier::EnvironmentOnly, false, elsewhere)
.starts_with("this host has no kernel containment for capsule commands"));
assert!(reason(EnforcementTier::KernelSeccompOnly, false, elsewhere)
.contains("neither a private mount namespace nor Landlock"));
assert!(reason(EnforcementTier::KernelFull, false, elsewhere)
.contains("a literal /tmp path is refused"));
assert!(reason(EnforcementTier::KernelFull, true, tmp.as_path())
.starts_with("the workdir is /tmp itself"));
}
#[test]
fn the_temp_folder_block_renders_after_the_records_directory() {
use crate::temp_dir::{NotMappedReason, TempMapping};
let placement = RuntimeDirPlacement::outside_for_test();
let mapped = temp_dir_report(TempMapping::MountNamespace, &placement);
assert_eq!(
render_temp_dir(&mapped),
"Temporary folder\n path: /home/test/.murmur/records/work/<session-id>/tmp\n \
TMPDIR: /tmp\n /tmp: mapped (mount-namespace)\n"
);
let unmapped = temp_dir_report(
TempMapping::NotMapped(NotMappedReason::NoMountNamespace),
&placement,
);
assert_eq!(
render_temp_dir(&unmapped),
format!(
"Temporary folder\n path: {path}\n TMPDIR: {path}\n /tmp: not \
mapped — {}\n",
NotMappedReason::NoMountNamespace.describe(),
path = "/home/test/.murmur/records/work/<session-id>/tmp"
)
);
assert_eq!(
serde_json::to_value(&mapped).unwrap(),
serde_json::json!({
"path": "/home/test/.murmur/records/work/<session-id>/tmp",
"tmpdir": "/tmp",
"tmp_mapped": true,
"mapped_by": "mount-namespace",
"not_mapped_because": null,
})
);
let scope = report_for(
&CapabilityPolicy::default(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
&placement,
TempMapping::MountNamespace,
IoMaxReport::default(),
);
let rendered = scope.render();
let records = rendered.find("Records directory\n").expect("records block");
let temp = rendered.find("Temporary folder\n").expect("temp block");
assert!(records < temp, "{rendered}");
assert!(rendered.contains(&render_temp_dir(&mapped)), "{rendered}");
assert!(serde_json::to_value(&scope)
.unwrap()
.get("temp_dir")
.is_some());
}
}