use super::facet::*;
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub struct OrdBound<T> {
pub min: Option<T>,
pub max: Option<T>,
}
impl<T: Ord + Copy> OrdBound<T> {
pub fn at_most(ceiling: T) -> Self {
Self { min: None, max: Some(ceiling) }
}
pub fn at_least(floor: T) -> Self {
Self { min: Some(floor), max: None }
}
pub fn exactly(exact: T) -> Self {
Self { min: Some(exact), max: Some(exact) }
}
pub fn admits(self, term: T) -> bool {
self.min.is_none_or(|lo| lo <= term) && self.max.is_none_or(|hi| term <= hi)
}
}
fn ord_admits<T: Ord + Copy>(bound: Option<OrdBound<T>>, term: T) -> bool {
bound.is_none_or(|b| b.admits(term))
}
fn set_admits<T: Eq + Copy>(set: Option<&[T]>, term: T) -> bool {
set.is_none_or(|s| s.contains(&term))
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FacetMismatch {
pub facet: &'static str,
pub actual: &'static str,
pub bound: String,
pub admits_count: usize,
}
impl std::fmt::Display for FacetMismatch {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{} = {} (allowed: {})", self.facet, self.actual, self.bound)
}
}
fn ord_mismatch<T: Ord + Copy + FacetTerm>(
facet: &'static str,
bound: Option<OrdBound<T>>,
term: T,
) -> Option<FacetMismatch> {
let b = bound?;
if b.admits(term) {
return None;
}
let bound = match (b.min, b.max) {
(None, Some(hi)) => format!("<= {}", hi.as_str()),
(Some(lo), None) => format!(">= {}", lo.as_str()),
(Some(lo), Some(hi)) => format!("{}..={}", lo.as_str(), hi.as_str()),
(None, None) => "any".to_string(),
};
let admits_count = T::all().iter().filter(|t| b.admits(**t)).count();
Some(FacetMismatch { facet, actual: term.as_str(), bound, admits_count })
}
fn set_mismatch<T: PartialEq + Copy + FacetTerm>(
facet: &'static str,
set: Option<&[T]>,
term: T,
) -> Option<FacetMismatch> {
let s = set?;
if s.contains(&term) {
return None;
}
let bound = format!(
"one of [{}]",
s.iter().map(|t| t.as_str()).collect::<Vec<_>>().join(", "),
);
Some(FacetMismatch { facet, actual: term.as_str(), bound, admits_count: s.len() })
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Clause {
pub operation: Option<Vec<Operation>>,
pub local_locus: Option<OrdBound<LocalLocus>>,
pub remote_reach: Option<OrdBound<RemoteReach>>,
pub remote_binding: Option<Vec<RemoteBinding>>,
pub provenance: Option<OrdBound<Provenance>>,
pub scale: Option<OrdBound<Scale>>,
pub retrieval: Option<OrdBound<RetrievalGranularity>>,
pub authority: Option<OrdBound<Authority>>,
pub isolation: Option<OrdBound<Isolation>>,
pub reversibility: Option<OrdBound<Reversibility>>,
pub persistence_level: Option<OrdBound<PersistenceLevel>>,
pub trigger_escape: Option<OrdBound<TriggerEscape>>,
pub trigger_kind: Option<Vec<TriggerKind>>,
pub disclosure_audience: Option<OrdBound<DisclosureAudience>>,
pub disclosure_channel: Option<Vec<Channel>>,
pub disclosure_principal: Option<Vec<Principal>>,
pub secret_level: Option<OrdBound<SecretLevel>>,
pub secret_channel: Option<Vec<Channel>>,
pub secret_principal: Option<Vec<Principal>>,
pub net_direction: Option<OrdBound<NetDirection>>,
pub net_destination: Option<OrdBound<NetDestination>>,
pub net_payload: Option<OrdBound<NetPayload>>,
pub execution_trust: Option<OrdBound<ExecutionTrust>>,
pub supply_source: Option<Vec<SupplySource>>,
pub pinning: Option<OrdBound<Pinning>>,
pub exec_surface: Option<Vec<ExecSurface>>,
pub cost: Option<OrdBound<Cost>>,
}
impl Clause {
pub fn admits(&self, cap: &Capability) -> bool {
self.check(cap, Role::Allow)
}
fn matches_as_deny(&self, cap: &Capability) -> bool {
self.check(cap, Role::Deny)
}
fn check(&self, cap: &Capability, role: Role) -> bool {
self.first_mismatch(cap, role).is_none()
}
#[cfg(test)]
pub(crate) fn first_mismatch_for_test(
&self,
cap: &Capability,
deny_role: bool,
) -> Option<FacetMismatch> {
self.first_mismatch(cap, if deny_role { Role::Deny } else { Role::Allow })
}
#[cfg(test)]
pub(crate) fn matches_as_deny_for_test(&self, cap: &Capability) -> bool {
self.matches_as_deny(cap)
}
fn first_mismatch(&self, cap: &Capability, role: Role) -> Option<FacetMismatch> {
set_mismatch("operation", self.operation.as_deref(), cap.operation)
.or_else(|| ord_mismatch("locus.local", self.local_locus, cap.locus.local))
.or_else(|| ord_mismatch("locus.remote", self.remote_reach, cap.locus.remote))
.or_else(|| set_mismatch("locus.binding", self.remote_binding.as_deref(), cap.locus.binding))
.or_else(|| ord_mismatch("locus.provenance", self.provenance, cap.locus.provenance))
.or_else(|| ord_mismatch("scale", self.scale, cap.scale))
.or_else(|| ord_mismatch("retrieval", self.retrieval, cap.retrieval))
.or_else(|| ord_mismatch("authority", self.authority, cap.authority))
.or_else(|| ord_mismatch("isolation", self.isolation, cap.isolation))
.or_else(|| ord_mismatch("reversibility", self.reversibility, cap.reversibility))
.or_else(|| ord_mismatch("persistence.level", self.persistence_level, cap.persistence.level))
.or_else(|| ord_mismatch("persistence.trigger.escape", self.trigger_escape, cap.persistence.trigger.escape))
.or_else(|| set_mismatch("persistence.trigger.kind", self.trigger_kind.as_deref(), cap.persistence.trigger.kind))
.or_else(|| ord_mismatch("disclosure.audience", self.disclosure_audience, cap.disclosure.audience))
.or_else(|| set_mismatch("disclosure.channel", self.disclosure_channel.as_deref(), cap.disclosure.channel))
.or_else(|| set_mismatch("disclosure.principal", self.disclosure_principal.as_deref(), cap.disclosure.principal))
.or_else(|| ord_mismatch("secret.level", self.secret_level, cap.secret.level))
.or_else(|| set_mismatch("secret.channel", self.secret_channel.as_deref(), cap.secret.channel))
.or_else(|| set_mismatch("secret.principal", self.secret_principal.as_deref(), cap.secret.principal))
.or_else(|| ord_mismatch("network.direction", self.net_direction, cap.network.direction))
.or_else(|| ord_mismatch("network.destination", self.net_destination, cap.network.destination))
.or_else(|| ord_mismatch("network.payload", self.net_payload, cap.network.payload))
.or_else(|| ord_mismatch("execution.trust", self.execution_trust, cap.execution.trust))
.or_else(|| self.supply_chain_mismatch(cap.execution.supply_chain, role))
.or_else(|| ord_mismatch("cost", self.cost, cap.cost))
}
fn supply_chain_mismatch(&self, sc: Option<SupplyChain>, role: Role) -> Option<FacetMismatch> {
if self.supply_chain_admits(sc, role) {
return None;
}
let Some(sc) = sc else {
return Some(FacetMismatch {
facet: "execution.supply_chain",
actual: "absent",
bound: "this clause constrains the supply chain, which this capability has none of"
.to_string(),
admits_count: 0,
});
};
set_mismatch("supply_chain.source", self.supply_source.as_deref(), sc.source)
.or_else(|| ord_mismatch("supply_chain.pinning", self.pinning, sc.pinning))
.or_else(|| set_mismatch("supply_chain.exec_surface", self.exec_surface.as_deref(), sc.exec_surface))
}
fn supply_chain_admits(&self, sc: Option<SupplyChain>, role: Role) -> bool {
match sc {
None => match role {
Role::Allow => true,
Role::Deny => {
self.supply_source.is_none()
&& self.pinning.is_none()
&& self.exec_surface.is_none()
}
},
Some(sc) => {
set_admits(self.supply_source.as_deref(), sc.source)
&& ord_admits(self.pinning, sc.pinning)
&& set_admits(self.exec_surface.as_deref(), sc.exec_surface)
}
}
}
}
#[derive(Copy, Clone, PartialEq, Eq)]
enum Role {
Allow,
Deny,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Level {
pub name: String,
pub allow: Vec<Clause>,
pub deny: Vec<Clause>,
}
impl Level {
pub fn new(name: impl Into<String>) -> Self {
Self { name: name.into(), allow: Vec::new(), deny: Vec::new() }
}
#[must_use]
pub fn allowing(mut self, clause: Clause) -> Self {
self.allow.push(clause);
self
}
#[must_use]
pub fn denying(mut self, clause: Clause) -> Self {
self.deny.push(clause);
self
}
pub fn nearest_miss(&self, cap: &Capability) -> Option<FacetMismatch> {
if self.allow.iter().any(|c| c.admits(cap)) {
return self.deny.iter().find(|c| c.matches_as_deny(cap)).map(|_| FacetMismatch {
facet: "deny-clause",
actual: "matched",
bound: "removed by an explicit deny clause".to_string(),
admits_count: 0,
});
}
if self.allow.is_empty() {
return Some(FacetMismatch {
facet: "level",
actual: "any capability",
bound: "nothing — this level declares no allow clause".to_string(),
admits_count: 0,
});
}
let on_topic = |c: &&Clause| {
c.operation.as_ref().is_none_or(|ops| ops.contains(&cap.operation))
};
let mut candidates: Vec<FacetMismatch> = self
.allow
.iter()
.filter(on_topic)
.filter_map(|c| c.first_mismatch(cap, Role::Allow))
.collect();
if candidates.is_empty() {
candidates = self
.allow
.iter()
.filter_map(|c| c.first_mismatch(cap, Role::Allow))
.collect();
}
candidates.into_iter().max_by_key(|m| m.admits_count)
}
pub fn admits_capability(&self, cap: &Capability) -> bool {
self.allow.iter().any(|c| c.admits(cap)) && !self.deny.iter().any(|c| c.matches_as_deny(cap))
}
pub fn admits(&self, profile: &Profile) -> bool {
profile.capabilities.iter().all(|c| self.admits_capability(c))
}
#[must_use]
pub fn extend(base: &Level, name: impl Into<String>, extra_allow: Vec<Clause>) -> Level {
let mut allow = base.allow.clone();
allow.extend(extra_allow);
Level { name: name.into(), allow, deny: base.deny.clone() }
}
}
#[cfg(test)]
mod tests {
use super::*;
use proptest::prelude::*;
fn cap(op: Operation) -> Capability {
Capability::new(op)
}
#[test]
fn empty_clause_admits_everything_empty_allow_admits_nothing() {
let all = Level::new("all").allowing(Clause::default());
let nothing = Level::new("nothing");
let destroy = Profile::of(vec![cap(Operation::Destroy)]);
assert!(all.admits(&destroy));
assert!(!nothing.admits(&destroy));
assert!(all.admits(&Profile::default()));
assert!(nothing.admits(&Profile::default()));
}
#[test]
fn nearest_miss_reports_the_clause_that_came_closest_not_the_first() {
let strict = Clause {
operation: Some(vec![Operation::Observe]),
local_locus: Some(OrdBound::at_most(LocalLocus::Temp)),
..Default::default()
};
let loose = Clause {
operation: Some(vec![Operation::Observe]),
local_locus: Some(OrdBound::at_most(LocalLocus::WorktreeTrusted)),
..Default::default()
};
let level = Level::new("two-reads").allowing(strict).allowing(loose);
let mut cap = Capability::new(Operation::Observe);
cap.locus.local = LocalLocus::Machine;
let m = level.nearest_miss(&cap).expect("machine locus exceeds both clauses");
assert_eq!(m.facet, "locus.local");
assert!(
m.bound.contains("worktree-trusted"),
"named the strictest clause (`{}`); the closest one allows <= worktree-trusted",
m.bound,
);
}
#[test]
fn read_local_admits_a_read_rejects_a_secret_and_a_destroy() {
let read_local = Level::new("read-local").allowing(Clause {
operation: Some(vec![Operation::Observe]),
local_locus: Some(OrdBound::at_most(LocalLocus::User)),
secret_level: Some(OrdBound::at_most(SecretLevel::UsesAmbient)),
net_direction: Some(OrdBound::at_most(NetDirection::Loopback)),
..Default::default()
});
let plain_read = Profile::of(vec![{
let mut c = cap(Operation::Observe);
c.locus.local = LocalLocus::Worktree;
c
}]);
assert!(read_local.admits(&plain_read));
let secret_read = Profile::of(vec![{
let mut c = cap(Operation::Observe);
c.locus.local = LocalLocus::User;
c.secret.level = SecretLevel::Reads;
c
}]);
assert!(!read_local.admits(&secret_read), "cat ~/.ssh/id_rsa must not pass read-local");
let destroy = Profile::of(vec![cap(Operation::Destroy)]);
assert!(!read_local.admits(&destroy));
}
#[test]
fn yolo_deny_carves_out_the_catastrophe_corner() {
let yolo = Level::new("yolo").allowing(Clause::default()).denying(Clause {
operation: Some(vec![Operation::Destroy]),
reversibility: Some(OrdBound::at_least(Reversibility::Irreversible)),
scale: Some(OrdBound::at_least(Scale::Unbounded)),
..Default::default()
});
let bounded = Profile::of(vec![{
let mut c = cap(Operation::Destroy);
c.scale = Scale::Bounded;
c.reversibility = Reversibility::Recoverable;
c
}]);
assert!(yolo.admits(&bounded));
let wipe = Profile::of(vec![{
let mut c = cap(Operation::Destroy);
c.scale = Scale::Unbounded;
c.reversibility = Reversibility::Irreversible;
c
}]);
assert!(!yolo.admits(&wipe));
}
#[test]
fn supply_chain_gates_network_sourced_code_and_is_vacuous_otherwise() {
let dev = Level::new("dev").allowing(Clause {
execution_trust: Some(OrdBound::at_most(ExecutionTrust::NetworkSourced)),
supply_source: Some(vec![
SupplySource::PublicRegistry,
SupplySource::SignedRepo,
SupplySource::PrivateRegistry,
SupplySource::Vendored,
]),
..Default::default()
});
let build = Profile::of(vec![{
let mut c = cap(Operation::Execute);
c.execution = Execution {
trust: ExecutionTrust::NetworkSourced,
supply_chain: Some(SupplyChain {
source: SupplySource::PublicRegistry,
pinning: Pinning::HashVerified,
exec_surface: ExecSurface::BuildScript,
}),
};
c
}]);
assert!(dev.admits(&build), "cargo build from a registry");
let curl_sh = Profile::of(vec![{
let mut c = cap(Operation::Execute);
c.execution = Execution {
trust: ExecutionTrust::NetworkSourced,
supply_chain: Some(SupplyChain {
source: SupplySource::UnverifiedUrl,
pinning: Pinning::Floating,
exec_surface: ExecSurface::RunArtifact,
}),
};
c
}]);
assert!(!dev.admits(&curl_sh), "curl | sh is an unverified-url source");
let plain = Profile::of(vec![cap(Operation::Observe)]);
assert!(dev.admits(&plain), "a command with no supply chain passes vacuously");
}
#[test]
fn a_supply_chain_deny_does_not_match_a_capability_without_one() {
let level = Level::new("x").allowing(Clause::default()).denying(Clause {
supply_source: Some(vec![SupplySource::UnverifiedUrl]),
..Default::default()
});
let plain = Profile::of(vec![cap(Operation::Observe)]);
assert!(level.admits(&plain), "a supply-chain deny must not match a no-supply-chain cap");
let curl_sh = Profile::of(vec![{
let mut c = cap(Operation::Execute);
c.execution = Execution {
trust: ExecutionTrust::NetworkSourced,
supply_chain: Some(SupplyChain {
source: SupplySource::UnverifiedUrl,
pinning: Pinning::Floating,
exec_surface: ExecSurface::RunArtifact,
}),
};
c
}]);
assert!(!level.admits(&curl_sh), "the deny corner still catches the unverified-url source");
}
#[test]
fn extend_inherits_deny_and_adds_allow() {
let base = Level::new("base")
.allowing(Clause { operation: Some(vec![Operation::Observe]), ..Default::default() })
.denying(Clause {
local_locus: Some(OrdBound::at_least(LocalLocus::Device)),
..Default::default()
});
let child = Level::extend(
&base,
"child",
vec![Clause {
operation: Some(vec![Operation::Create, Operation::Mutate]),
..Default::default()
}],
);
assert!(child.admits(&Profile::of(vec![cap(Operation::Create)])), "added allow");
assert!(child.admits(&Profile::of(vec![cap(Operation::Observe)])), "inherited allow");
let device = Profile::of(vec![{
let mut c = cap(Operation::Mutate);
c.locus.local = LocalLocus::Device;
c
}]);
assert!(!child.admits(&device), "inherited deny still bites");
}
use crate::engine::testgen::{arb_clause, arb_level, arb_profile};
proptest! {
#[test]
fn totality(level in arb_level(), profile in arb_profile()) {
let first = level.admits(&profile);
prop_assert_eq!(first, level.admits(&profile));
}
#[test]
fn extends_is_a_superset(
base in arb_level(),
extra in prop::collection::vec(arb_clause(), 0..3),
profile in arb_profile(),
) {
let extended = Level::extend(&base, "child", extra);
prop_assert!(!base.admits(&profile) || extended.admits(&profile));
}
#[test]
fn deny_only_shrinks(
level in arb_level(),
extra_deny in arb_clause(),
profile in arb_profile(),
) {
let stricter = level.clone().denying(extra_deny);
prop_assert!(!stricter.admits(&profile) || level.admits(&profile));
}
}
}