use murmur_artifact::{ContainmentClass, ExportMode, Exports, FileExport, PeerFilesExport};
use serde::Serialize;
use crate::{
cgroup::{IoMaxReport, IoMaxStatus},
errors::RuntimeError,
sandbox::{detect_enforcement_tier, EnforcementTier},
sealed::{SealedBlocker, UsernsGrant},
types::CapabilityPolicy,
};
pub use crate::sandbox::{detect_sealed_blocker, 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
}
}
}
pub(crate) fn achieved_containment_class(
tier: EnforcementTier,
workdir_exec: bool,
) -> ContainmentClass {
let host_ceiling = achieved_class_for_tier(tier);
if workdir_exec {
host_ceiling.min(ContainmentClass::Advisory)
} else {
host_ceiling
}
}
pub fn detect_achieved_containment() -> ContainmentClass {
achieved_class_for_tier(detect_enforcement_tier())
}
pub fn containment_shortfall_reason(
declared: ContainmentClass,
achieved: ContainmentClass,
sealed_blocker: Option<SealedBlocker>,
workdir_exec: bool,
) -> Option<String> {
if achieved >= declared {
return None;
}
if workdir_exec && declared > ContainmentClass::Advisory {
return Some(WORKDIR_EXEC_SHORTFALL.to_string());
}
let reason = match declared {
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_blocker {
Some(blocker) => blocker.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)
}
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_blocker: Option<SealedBlocker>,
workdir_exec: bool,
) -> Result<(), RuntimeError> {
match containment_shortfall_reason(declared, achieved, sealed_blocker, workdir_exec) {
None => Ok(()),
Some(reason) => Err(RuntimeError::ContainmentFloorUnmet {
declared,
achieved,
reason,
}),
}
}
#[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,
}
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",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct PreopenReport {
pub artifact: String,
pub role: String,
pub scope: Option<String>,
pub surface: PreopenSurface,
}
impl PreopenReport {
pub fn render(&self) -> String {
let artifact = &self.artifact;
let role = &self.role;
let surface = self.surface.summary();
match &self.scope {
Some(scope) => {
format!("{artifact} ({role}): {surface} — capabilities.filesystem.scope: {scope}")
}
None => format!(
"{artifact} ({role}): {surface} — no capabilities.filesystem.scope declared"
),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum FilesystemRestriction {
Advisory,
Enforced,
Absent,
}
#[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 session workdir, 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 PATH_REWRITING_NOT_PROTECTED: &str =
"Path rewriting is a convenience, not a boundary: HOME is set to <workdir>/.capsule-home on \
every tier, so `cat ~/.ssh/id_rsa` answers \"No such file or directory\" whether or not \
anything was denied — the tilde resolved into the workdir 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) fn filesystem_boundary(applied: EnforcementTier) -> FilesystemBoundaryReport {
let (restriction, filesystem) = match applied {
EnforcementTier::KernelSealed => (FilesystemRestriction::Absent, None),
EnforcementTier::KernelFull => (
FilesystemRestriction::Enforced,
Some(ENFORCED_FILESYSTEM_NOT_PROTECTED),
),
EnforcementTier::KernelSeccompOnly | EnforcementTier::EnvironmentOnly => (
FilesystemRestriction::Advisory,
Some(ADVISORY_FILESYSTEM_NOT_PROTECTED),
),
};
let not_protected = match filesystem {
Some(filesystem) => vec![
filesystem.to_string(),
PATH_REWRITING_NOT_PROTECTED.to_string(),
],
None => Vec::new(),
};
FilesystemBoundaryReport {
restriction,
not_protected,
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ScopeReport {
pub declared_containment: ContainmentClass,
pub achieved_containment: ContainmentClass,
pub floor_met: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub shortfall_reason: Option<String>,
pub enforcement_tier: &'static str,
pub userns_grant: Option<UsernsGrant>,
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 env_allow: Vec<String>,
pub plan_submit: bool,
pub interpreter_runtime_grants: Vec<String>,
pub exports_files: Option<ExportsFilesReport>,
pub peer_files: Option<PeerFilesReport>,
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 io_max: IoMaxReport,
pub filesystem_boundary: FilesystemBoundaryReport,
}
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!(
" 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(&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, "env allow", &self.env_allow);
out.push_str(&format!(" plan submit: {}\n", self.plan_submit));
push_list(
&mut out,
"interpreter runtime",
&self.interpreter_runtime_grants,
);
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);
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
));
}
}
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,
));
if !self.floor_met {
out.push_str(
"\nThis is a report only — `mur run` without --explain-scope would refuse to \
launch here.\n",
);
}
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
}
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"));
}
}
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,
) -> ScopeReport {
let probe = crate::sandbox::HostProbe::probe();
scope_report_for_tier(
policy,
declared,
probe.tier(),
probe.sealed_blocker(),
probe.userns_grant(),
exports,
state_stores,
configured_artifacts,
preopens,
io_max,
)
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn scope_report_for_tier(
policy: &CapabilityPolicy,
declared: ContainmentClass,
tier: EnforcementTier,
sealed_blocker: Option<SealedBlocker>,
userns_grant: Option<UsernsGrant>,
exports: Option<&Exports>,
state_stores: Vec<StateStoreReport>,
configured_artifacts: Vec<String>,
preopens: Vec<PreopenReport>,
io_max: IoMaxReport,
) -> ScopeReport {
let achieved = achieved_containment_class(tier, policy.workdir_exec_allowed);
let applied = crate::sandbox::applied_tier(tier, declared);
let shortfall_reason = containment_shortfall_reason(
declared,
achieved,
sealed_blocker,
policy.workdir_exec_allowed,
);
ScopeReport {
declared_containment: declared,
achieved_containment: achieved,
floor_met: shortfall_reason.is_none(),
shortfall_reason,
enforcement_tier: enforcement_tier_name(tier),
userns_grant,
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_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(),
env_allow: policy.env_allow.clone(),
plan_submit: policy.plan_submit,
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(),
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(applied),
io_max,
filesystem_boundary: filesystem_boundary(applied),
}
}
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",
}
}
#[cfg(test)]
mod tests {
use super::*;
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),
ContainmentClass::Sealed
);
assert_eq!(
achieved_containment_class(EnforcementTier::KernelFull, false),
ContainmentClass::Scoped
);
assert_eq!(
achieved_containment_class(EnforcementTier::KernelSeccompOnly, false),
ContainmentClass::Advisory
);
assert_eq!(
achieved_containment_class(EnforcementTier::EnvironmentOnly, false),
ContainmentClass::Advisory
);
for tier in ALL_TIERS {
assert_eq!(
achieved_containment_class(*tier, true),
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),
"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);
assert_eq!(achieved, ContainmentClass::Advisory);
let error = check_containment_floor(ContainmentClass::Scoped, achieved, None, true)
.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(SealedBlocker::AppArmorProfileMissing),
true,
)
.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);
assert!(
check_containment_floor(ContainmentClass::Advisory, achieved, None, true).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).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).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)
.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() {
let achieved = achieved_class_for_tier(EnforcementTier::KernelSealed);
assert_eq!(achieved, ContainmentClass::Sealed);
assert!(check_containment_floor(ContainmentClass::Advisory, achieved, None, false).is_ok());
assert!(check_containment_floor(ContainmentClass::Scoped, achieved, None, false).is_ok());
assert_eq!(
containment_shortfall_reason(ContainmentClass::Scoped, achieved, None, false),
None
);
}
#[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
)
.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(SealedBlocker::NamespaceCreationDenied),
false,
)
.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("--cap-add SYS_ADMIN"),
"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(SealedBlocker::AppArmorProfileMissing),
false,
)
.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(SealedBlocker::NamespaceCreationDenied),
false,
)
.expect("refusal");
assert!(container.contains("--cap-add SYS_ADMIN"));
assert!(container.contains("outside the container"));
assert!(!container.contains("apparmor_parser"));
let pivot = containment_shortfall_reason(
ContainmentClass::Sealed,
ContainmentClass::Advisory,
Some(SealedBlocker::PivotRootDenied),
false,
)
.expect("refusal");
assert!(pivot.contains("pivot_root(2)"), "reason was: {pivot}");
assert_ne!(
pivot,
containment_shortfall_reason(
ContainmentClass::Sealed,
ContainmentClass::Advisory,
Some(SealedBlocker::MountDenied),
false,
)
.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(blocker),
false,
)
.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,
)
.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
);
assert_eq!(
containment_shortfall_reason(
ContainmentClass::Scoped,
ContainmentClass::Scoped,
None,
false
),
None
);
assert_eq!(
containment_shortfall_reason(
ContainmentClass::Sealed,
ContainmentClass::Sealed,
Some(SealedBlocker::NamespaceCreationDenied),
false
),
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 scope_report_mirrors_the_policy_and_the_probed_tier() {
let report = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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(scope_report_for_tier(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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(scope_report_for_tier(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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(scope_report_for_tier(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&policy,
ContainmentClass::Sealed,
*tier,
Some(SealedBlocker::NamespaceCreationDenied),
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::KernelFull,
Some(SealedBlocker::NamespaceCreationDenied),
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::EnvironmentOnly,
Some(SealedBlocker::NotLinux),
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::KernelSealed,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
);
assert_eq!(report.achieved_containment, ContainmentClass::Sealed);
assert!(report.floor_met);
assert_eq!(
report.enforcement_tier,
"mountns+pivot_root+landlock+seccomp"
);
}
#[test]
fn scope_report_json_uses_wire_names() {
let report = scope_report_for_tier(
&CapabilityPolicy::default(),
ContainmentClass::Advisory,
EnforcementTier::KernelSeccompOnly,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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["floor_met"], true);
assert_eq!(value["enforcement_tier"], "seccomp-only");
assert!(value.get("shortfall_reason").is_none());
}
#[test]
fn the_userns_grant_is_reported_without_changing_any_verdict() {
let baseline = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
);
for grant in UsernsGrant::ALL {
let report = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
Some(*grant),
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::KernelFull,
Some(SealedBlocker::AppArmorProfileMissing),
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&policy,
ContainmentClass::Scoped,
*tier,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
);
let with = scope_report_for_tier(
&policy,
ContainmentClass::Scoped,
*tier,
None,
None,
Some(&Exports {
files: Some(export.clone()),
peer_files: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&without_policy,
ContainmentClass::Scoped,
*tier,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
);
let with = scope_report_for_tier(
&with_policy,
ContainmentClass::Scoped,
*tier,
None,
None,
Some(&Exports {
files: None,
peer_files: Some(peer_files.clone()),
}),
Vec::new(),
Vec::new(),
Vec::new(),
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(scope_report_for_tier(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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(scope_report_for_tier(
&CapabilityPolicy {
peer_fetch_allow: vec!["localhost:41234".to_string()],
..sample_policy()
},
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
Some(&Exports {
files: None,
peer_files: Some(PeerFilesExport {
root: "out/".to_string(),
max_ttl_secs: None,
max_bytes: 10 * 1024 * 1024,
}),
}),
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
)
.render();
assert!(undeclared.contains("Peer handoff"));
assert!(undeclared.contains("exports.peer_files: <none>"));
assert!(undeclared.contains("peer_fetch allow: <none>"));
let declared = scope_report_for_tier(
&CapabilityPolicy {
peer_fetch_allow: vec!["localhost:41234".to_string()],
..sample_policy()
},
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
Some(&Exports {
files: None,
peer_files: Some(PeerFilesExport {
root: "out/".to_string(),
max_ttl_secs: Some(900),
max_bytes: 4096,
}),
}),
Vec::new(),
Vec::new(),
Vec::new(),
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"));
}
#[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 = scope_report_for_tier(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
);
let with = scope_report_for_tier(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
stores.clone(),
Vec::new(),
Vec::new(),
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",
}])
);
}
#[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,
},
PreopenReport {
artifact: "cache-tool".to_string(),
role: "tool".to_string(),
scope: Some("cache".to_string()),
surface: PreopenSurface::ScopedSubtree,
},
PreopenReport {
artifact: "telemetry-hook".to_string(),
role: "hook".to_string(),
scope: None,
surface: PreopenSurface::Nothing,
},
];
let without = scope_report_for_tier(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
);
let with = scope_report_for_tier(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
preopens.clone(),
IoMaxReport::default(),
);
assert_eq!(with.preopens, preopens);
assert!(without.preopens.is_empty());
assert_eq!(
ScopeReport {
preopens: Vec::new(),
..with.clone()
},
without
);
let rendered = with.render();
assert!(rendered.contains(
"- notes-tool (tool): the whole accessible workdir — no capabilities.filesystem.scope \
declared"
));
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",
},
{
"artifact": "cache-tool",
"role": "tool",
"scope": "cache",
"surface": "scoped-subtree",
},
{
"artifact": "telemetry-hook",
"role": "hook",
"scope": null,
"surface": "nothing",
},
])
);
}
#[test]
fn exports_files_is_always_serialized() {
let undeclared = serde_json::to_value(scope_report_for_tier(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
))
.unwrap();
assert_eq!(undeclared["exports_files"], serde_json::Value::Null);
let declared = serde_json::to_value(scope_report_for_tier(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
Some(&Exports {
files: Some(FileExport {
root: "out/".to_string(),
mode: ExportMode::ReadOnly,
max_bytes: 10_485_760,
}),
peer_files: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
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 = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
)
.render();
assert!(
undeclared.contains("Resource plane") && undeclared.contains("exports.files: <none>"),
"{undeclared}"
);
let declared = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Advisory,
EnforcementTier::KernelFull,
None,
None,
Some(&Exports {
files: Some(FileExport {
root: "out/".to_string(),
mode: ExportMode::ReadOnly,
max_bytes: 10_485_760,
}),
peer_files: None,
}),
Vec::new(),
Vec::new(),
Vec::new(),
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 report_for(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()
};
scope_report_for_tier(
&policy,
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
)
}
#[test]
fn declared_read_only_with_no_interpreter_reports_as_enforced() {
let report = report_for(&["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 = report_for(&["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!(
report_for(&["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(report_for(&[], &["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(report_for(&["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 = report_for(&[], &["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 native 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"),
("shell_interpreter_runtime", "interpreter_runtime_grants"),
("shell_staged_runtime", "staged_runtime_grants"),
("env_allow", "env_allow"),
("plan_submit", "plan_submit"),
("containment_floor", "declared_containment"),
];
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(scope_report_for_tier(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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 {
scope_report_for_tier(
&sample_policy(),
declared,
tier,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
)
}
#[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(), 2);
let filesystem = &report.filesystem_boundary.not_protected[0];
assert!(filesystem.contains("advisory"), "{filesystem}");
assert!(
filesystem.contains("denies a read or a write outside the session workdir"),
"{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}");
}
}
#[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::Sealed,
"the host can back sealed, which is what `achieved` answers"
);
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(crate::shell::SYNTHETIC_HOME_DIR_NAME),
"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 = scope_report_for_tier(
&policy,
ContainmentClass::Advisory,
EnforcementTier::KernelSealed,
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
IoMaxReport::default(),
);
assert_eq!(
report.enforcement_tier,
"mountns+pivot_root+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
}
};
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",
};
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 enforcement =
crate::sandbox::ShellEnforcement::resolve(&policy, declared, probe).unwrap();
let report = scope_report_for_tier(
&policy,
declared,
probe.tier(),
None,
None,
None,
Vec::new(),
Vec::new(),
Vec::new(),
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],
&[],
workdir.path(),
workdir.path(),
&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}"
),
}
}
}
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()
}
}