use murmur_artifact::ContainmentClass;
use serde::Serialize;
use crate::{
errors::RuntimeError,
sandbox::{detect_enforcement_tier, EnforcementTier},
sealed::SealedBlocker,
types::CapabilityPolicy,
};
pub use crate::sandbox::detect_sealed_blocker;
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 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 filesystem_scope: Option<String>,
pub workdir_exec: bool,
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 interpreter_runtime_grants: Vec<String>,
pub staged_runtime_grants: Vec<String>,
}
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("\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));
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);
push_list(
&mut out,
"interpreter runtime",
&self.interpreter_runtime_grants,
);
push_list(&mut out, "staged runtime", &self.staged_runtime_grants);
if !self.floor_met {
out.push_str(
"\nThis is a report only — `mur run` without --explain-scope would refuse to \
launch here.\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) -> ScopeReport {
scope_report_for_tier(
policy,
declared,
detect_enforcement_tier(),
detect_sealed_blocker(),
)
}
pub(crate) fn scope_report_for_tier(
policy: &CapabilityPolicy,
declared: ContainmentClass,
tier: EnforcementTier,
sealed_blocker: Option<SealedBlocker>,
) -> ScopeReport {
let achieved = achieved_containment_class(tier, policy.workdir_exec_allowed);
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),
filesystem_scope: policy.filesystem_scope.clone(),
workdir_exec: policy.workdir_exec_allowed,
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(),
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(),
}
}
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"));
for blocker in [
SealedBlocker::NotLinux,
SealedBlocker::AppArmorProfileMissing,
SealedBlocker::NamespaceCreationDenied,
SealedBlocker::MountDenied,
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,
);
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,
);
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_workdir_exec() {
let value: serde_json::Value = serde_json::to_value(scope_report_for_tier(
&sample_policy(),
ContainmentClass::Scoped,
EnforcementTier::KernelFull,
None,
))
.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),
);
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,
);
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),
);
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),
);
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,
);
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,
);
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 rendered_report_names_both_classes_and_the_refusal() {
let rendered = scope_report_for_tier(
&sample_policy(),
ContainmentClass::Sealed,
EnforcementTier::KernelFull,
Some(SealedBlocker::AppArmorProfileMissing),
)
.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"));
}
}