#![cfg_attr(
not(test),
deny(
clippy::unwrap_used,
clippy::expect_used,
clippy::panic,
clippy::unreachable,
clippy::todo,
clippy::unimplemented,
clippy::indexing_slicing,
clippy::string_slice,
clippy::arithmetic_side_effects,
)
)]
pub mod audit;
pub mod db;
pub mod jobs;
pub mod kv;
pub mod outbound;
pub mod quota;
pub mod render;
use std::collections::BTreeMap;
use std::fmt;
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use super::grants::CapabilityQuotas;
use super::manifest::{SandboxCapability, SandboxManifest};
use quota::QuotaLedger;
pub use audit::{
ActivitySummary, CapabilityEvent, CapabilityOutcome, DroppedCalls, MAX_LOG_EVENTS,
PluginActivityLog,
};
use audit::{CapabilityEvent as Event, CapabilityOutcome as Outcome};
pub use db::{MemoryPluginStore, PluginStore, QueryPage, Scope, StoreError};
pub use jobs::MemoryJobSink;
pub use jobs::{JobSink, PluginJob};
pub use kv::{CacheKvStore, KvStore, MemoryKvStore};
pub use outbound::{
ALLOWED_OUTBOUND_REQUEST_HEADERS, ALLOWED_OUTBOUND_RESPONSE_HEADERS, MAX_OUTBOUND_HEADERS,
MAX_RESPONSE_HEADER_BYTES, OutboundHttp, OutboundRequest, OutboundResponse, RecordingHttp,
};
pub use quota::CapabilityRateLimiter;
pub use render::{ALLOWED_ATTRIBUTES, ALLOWED_TAGS, FragmentNode, RenderError};
pub const MAX_ROW_COLUMNS: usize = 32;
pub const MAX_VALUE_TEXT_BYTES: usize = 64 * 1024;
pub const MAX_KV_KEY_BYTES: usize = 512;
pub const MAX_ROW_BYTES: usize = 256 * 1024;
pub const MAX_RESULT_BYTES: usize = 512 * 1024;
pub const MAX_ROW_ID_BYTES: usize = 128;
pub const MAX_JOB_ID_BYTES: usize = 128;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum PluginValue {
Null,
Bool(bool),
Int(i64),
Float(f64),
Text(String),
}
thread_local! {
static FRAGMENT_NODE_BUDGET: std::cell::Cell<usize> =
const { std::cell::Cell::new(usize::MAX) };
}
pub struct FragmentNodeBudget(usize);
impl FragmentNodeBudget {
#[must_use]
pub fn reserve(limit: usize) -> Self {
Self(FRAGMENT_NODE_BUDGET.with(|budget| budget.replace(limit)))
}
}
impl Drop for FragmentNodeBudget {
fn drop(&mut self) {
FRAGMENT_NODE_BUDGET.with(|budget| budget.set(self.0));
}
}
pub fn bounded_tree<'de, D, const LIMIT: usize>(
deserializer: D,
) -> Result<Vec<render::FragmentNode>, D::Error>
where
D: serde::Deserializer<'de>,
{
let _budget = FragmentNodeBudget::reserve(LIMIT);
bounded_subtree(deserializer)
}
pub fn bounded_subtree<'de, D>(deserializer: D) -> Result<Vec<render::FragmentNode>, D::Error>
where
D: serde::Deserializer<'de>,
{
use serde::de::{Error as _, SeqAccess, Visitor};
struct Bounded;
impl<'de> Visitor<'de> for Bounded {
type Value = Vec<render::FragmentNode>;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
formatter,
"at most {} nodes in the whole fragment",
render::MAX_NODES
)
}
fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
let mut out = Vec::new();
while let Some(node) = seq.next_element()? {
let spent = FRAGMENT_NODE_BUDGET.with(|budget| {
let left = budget.get();
if left == 0 {
return false;
}
budget.set(left.saturating_sub(1));
true
});
if !spent {
return Err(A::Error::custom(format!(
"more than {} nodes in the fragment",
render::MAX_NODES
)));
}
out.push(node);
}
Ok(out)
}
}
deserializer.deserialize_seq(Bounded)
}
pub fn bounded_vec<'de, D, T, const LIMIT: usize>(deserializer: D) -> Result<Vec<T>, D::Error>
where
D: serde::Deserializer<'de>,
T: serde::Deserialize<'de>,
{
use serde::de::{Error as _, SeqAccess, Visitor};
struct Bounded<T, const LIMIT: usize>(std::marker::PhantomData<T>);
impl<'de, T, const LIMIT: usize> Visitor<'de> for Bounded<T, LIMIT>
where
T: serde::Deserialize<'de>,
{
type Value = Vec<T>;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(formatter, "at most {LIMIT} elements")
}
fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
let mut out = Vec::with_capacity(seq.size_hint().unwrap_or(0).min(LIMIT));
while let Some(item) = seq.next_element()? {
if out.len() >= LIMIT {
return Err(A::Error::custom(format!("more than {LIMIT} elements")));
}
out.push(item);
}
Ok(out)
}
}
deserializer.deserialize_seq(Bounded::<T, LIMIT>(std::marker::PhantomData))
}
pub fn bounded_row<'de, D>(deserializer: D) -> Result<PluginRow, D::Error>
where
D: serde::Deserializer<'de>,
{
use serde::de::{Error as _, MapAccess, Visitor};
struct Bounded;
impl<'de> Visitor<'de> for Bounded {
type Value = PluginRow;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(formatter, "at most {} columns", MAX_ROW_COLUMNS + 1)
}
fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
const PARSE_CEILING: usize = MAX_ROW_COLUMNS + 1;
let mut out = PluginRow::new();
while let Some((column, value)) = map.next_entry::<String, PluginValue>()? {
if out.len() >= PARSE_CEILING && !out.contains_key(&column) {
return Err(A::Error::custom(format!(
"more than {PARSE_CEILING} columns"
)));
}
out.insert(column, value);
}
Ok(out)
}
}
deserializer.deserialize_map(Bounded)
}
#[must_use]
pub fn encoded_text_len(text: &str) -> usize {
text.chars()
.map(|ch| match ch {
'"' | '\\' | '\n' | '\r' | '\t' | '\u{8}' | '\u{c}' => 2,
other if other.is_control() => 6,
other => other.len_utf8(),
})
.fold(0, usize::saturating_add)
}
const fn decimal_len(value: i64) -> usize {
if value == i64::MIN {
return 20;
}
let negative = value < 0;
let mut left = value.unsigned_abs();
let mut digits = 1_usize;
while left >= 10 {
left /= 10;
digits = digits.saturating_add(1);
}
if negative {
digits.saturating_add(1)
} else {
digits
}
}
impl PluginValue {
#[must_use]
pub fn weight(&self) -> usize {
match self {
Self::Text(text) => encoded_text_len(text),
Self::Null | Self::Bool(true) => 4,
Self::Bool(false) => 5,
Self::Int(value) => decimal_len(*value),
Self::Float(_) => 24,
}
}
}
impl fmt::Display for PluginValue {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Null => f.write_str("null"),
Self::Bool(value) => write!(f, "{value}"),
Self::Int(value) => write!(f, "{value}"),
Self::Float(value) => write!(f, "{value}"),
Self::Text(value) => f.write_str(value),
}
}
}
pub type PluginRow = BTreeMap<String, PluginValue>;
pub fn check_row(row: &PluginRow) -> Result<(), String> {
check_row_without(row, None)
}
pub fn check_row_without(row: &PluginRow, skip: Option<&str>) -> Result<(), String> {
let kept = || {
row.iter()
.filter(move |(column, _)| Some(column.as_str()) != skip)
};
let columns = kept().count();
if columns > MAX_ROW_COLUMNS {
return Err(format!(
"a row may carry at most {MAX_ROW_COLUMNS} columns; this one carries {columns}"
));
}
let weight = kept().fold(0, |sum: usize, (column, value)| {
sum.saturating_add(column.len())
.saturating_add(value.weight())
});
if weight > MAX_ROW_BYTES {
return Err(format!(
"a row may carry at most {MAX_ROW_BYTES} bytes across its columns; this one carries \
{weight}"
));
}
for (column, value) in kept() {
if !super::grants::is_grantable_ident(column) {
return Err(format!(
"column name {:?} is not a lower-case `[a-z][a-z0-9_]*` identifier; the host \
derives a physical column name from it",
super::manifest::rejected(column)
));
}
if value.weight() > MAX_VALUE_TEXT_BYTES {
return Err(format!(
"column {column:?} carries {} bytes, over the {MAX_VALUE_TEXT_BYTES}-byte ceiling",
value.weight()
));
}
}
Ok(())
}
#[must_use]
pub fn row_weight(row: &PluginRow) -> usize {
row.iter().fold(0, |sum, (column, value)| {
sum.saturating_add(column.len())
.saturating_add(value.weight())
})
}
pub(super) fn row_framing(row: &PluginRow) -> usize {
row.len().saturating_mul(8).saturating_add(16)
}
#[must_use]
pub fn bounded_rows(rows: Vec<PluginRow>) -> (Vec<PluginRow>, bool) {
let mut kept = Vec::with_capacity(rows.len());
let mut total = 0_usize;
let mut truncated = false;
for row in rows {
let weight = row_weight(&row).saturating_add(row_framing(&row));
if total.saturating_add(weight) > MAX_RESULT_BYTES {
truncated = true;
break;
}
total = total.saturating_add(weight);
kept.push(row);
}
(kept, truncated)
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "call", rename_all = "kebab-case", deny_unknown_fields)]
#[non_exhaustive]
pub enum CapabilityCall {
KvGet {
id: u64,
key: String,
},
KvSet {
id: u64,
key: String,
value: PluginValue,
},
KvDelete {
id: u64,
key: String,
},
HttpFetch {
id: u64,
method: String,
url: String,
#[serde(
default,
deserialize_with = "bounded_vec::<_, _, MAX_OUTBOUND_HEADERS>"
)]
headers: Vec<(String, String)>,
#[serde(default)]
body: String,
},
DbInsert {
id: u64,
table: String,
#[serde(deserialize_with = "bounded_row")]
row: PluginRow,
},
DbGet {
id: u64,
table: String,
row_id: String,
},
DbQuery {
id: u64,
table: String,
#[serde(default, deserialize_with = "bounded_row")]
filter: PluginRow,
#[serde(default)]
limit: u32,
#[serde(default)]
after: Option<String>,
},
DbUpdate {
id: u64,
table: String,
row_id: String,
#[serde(deserialize_with = "bounded_row")]
row: PluginRow,
},
DbDelete {
id: u64,
table: String,
row_id: String,
},
JobEnqueue {
id: u64,
job_type: String,
#[serde(default, deserialize_with = "bounded_row")]
payload: PluginRow,
},
}
impl CapabilityCall {
#[must_use]
pub const fn id(&self) -> u64 {
match self {
Self::KvGet { id, .. }
| Self::KvSet { id, .. }
| Self::KvDelete { id, .. }
| Self::HttpFetch { id, .. }
| Self::DbInsert { id, .. }
| Self::DbGet { id, .. }
| Self::DbQuery { id, .. }
| Self::DbUpdate { id, .. }
| Self::DbDelete { id, .. }
| Self::JobEnqueue { id, .. } => *id,
}
}
#[must_use]
pub const fn capability(&self) -> SandboxCapability {
match self {
Self::KvGet { .. } | Self::KvSet { .. } | Self::KvDelete { .. } => {
SandboxCapability::Kv
}
Self::HttpFetch { .. } => SandboxCapability::HttpOutbound,
Self::DbInsert { .. }
| Self::DbGet { .. }
| Self::DbQuery { .. }
| Self::DbUpdate { .. }
| Self::DbDelete { .. } => SandboxCapability::Db,
Self::JobEnqueue { .. } => SandboxCapability::Jobs,
}
}
#[must_use]
pub fn target(&self) -> String {
match self {
Self::KvGet { key, .. } | Self::KvSet { key, .. } | Self::KvDelete { key, .. } => {
key.clone()
}
Self::HttpFetch { url, .. } => {
outbound::host_of(url).unwrap_or_else(|| outbound::attempted_authority(url))
}
Self::DbInsert { table, .. }
| Self::DbGet { table, .. }
| Self::DbQuery { table, .. }
| Self::DbUpdate { table, .. }
| Self::DbDelete { table, .. } => table.clone(),
Self::JobEnqueue { job_type, .. } => job_type.clone(),
}
}
#[must_use]
pub const fn reply_reserve(&self) -> usize {
self.reply_reserve_for(MAX_RESULT_BYTES, 500)
}
#[must_use]
pub const fn reply_reserve_for(&self, kv_value_bytes: usize, db_rows: usize) -> usize {
const ROW_FRAMING: usize = (MAX_ROW_COLUMNS + 1) * 8 + 16;
match self {
Self::KvGet { .. } => kv_value_bytes.saturating_add(4096),
Self::DbGet { .. } => MAX_RESULT_BYTES + ROW_FRAMING + 4096,
Self::DbQuery { limit, .. } => {
let asked = *limit as usize;
let rows = if asked == 0 || asked > db_rows {
db_rows
} else {
asked
};
let reserve = MAX_RESULT_BYTES
.saturating_add(rows.saturating_mul(ROW_FRAMING))
.saturating_add(4096);
if reserve > super::host::MAX_QUEUED_REPLY_BYTES {
super::host::MAX_QUEUED_REPLY_BYTES
} else {
reserve
}
}
Self::HttpFetch { .. } => {
MAX_RESULT_BYTES + outbound::MAX_RESPONSE_HEADER_BYTES * 6 + 4096
}
_ => 4096,
}
}
#[must_use]
pub fn target_excerpt(&self) -> String {
const URL_PREFIX_CHARS: usize = 4096;
match self {
Self::KvGet { key, .. } | Self::KvSet { key, .. } | Self::KvDelete { key, .. } => {
bounded_target(key)
}
Self::HttpFetch { url, .. } if url.len() <= URL_PREFIX_CHARS => bounded_target(
&outbound::host_of(url).unwrap_or_else(|| outbound::attempted_authority(url)),
),
Self::HttpFetch { url, .. } => {
let rest = url.split_once("://").map_or(url.as_str(), |(_, rest)| rest);
let rest = rest.trim_start_matches('/');
let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
let host = authority.rsplit('@').next().unwrap_or(authority);
bounded_target(host).to_ascii_lowercase()
}
Self::DbInsert { table, .. }
| Self::DbGet { table, .. }
| Self::DbQuery { table, .. }
| Self::DbUpdate { table, .. }
| Self::DbDelete { table, .. } => bounded_target(table),
Self::JobEnqueue { job_type, .. } => bounded_target(job_type),
}
}
#[must_use]
pub const fn operation(&self) -> &'static str {
match self {
Self::KvGet { .. } => "kv-get",
Self::KvSet { .. } => "kv-set",
Self::KvDelete { .. } => "kv-delete",
Self::HttpFetch { .. } => "http-fetch",
Self::DbInsert { .. } => "db-insert",
Self::DbGet { .. } => "db-get",
Self::DbQuery { .. } => "db-query",
Self::DbUpdate { .. } => "db-update",
Self::DbDelete { .. } => "db-delete",
Self::JobEnqueue { .. } => "job-enqueue",
}
}
#[must_use]
pub fn commits(&self) -> bool {
match self {
Self::KvGet { .. } | Self::DbGet { .. } | Self::DbQuery { .. } => false,
Self::HttpFetch { method, .. } => {
!(method.eq_ignore_ascii_case("GET") || method.eq_ignore_ascii_case("HEAD"))
}
Self::KvSet { .. }
| Self::KvDelete { .. }
| Self::DbInsert { .. }
| Self::DbUpdate { .. }
| Self::DbDelete { .. }
| Self::JobEnqueue { .. } => true,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
#[non_exhaustive]
pub enum DenialReason {
CapabilityNotGranted,
NotInGrant,
QuotaExceeded,
Malformed,
Unavailable,
BackendError,
ResponseTooLarge,
ResultRejected,
}
impl DenialReason {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::CapabilityNotGranted => "capability-not-granted",
Self::NotInGrant => "not-in-grant",
Self::QuotaExceeded => "quota-exceeded",
Self::Malformed => "malformed",
Self::Unavailable => "unavailable",
Self::BackendError => "backend-error",
Self::ResponseTooLarge => "response-too-large",
Self::ResultRejected => "result-rejected",
}
}
}
impl fmt::Display for DenialReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
#[must_use]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "kebab-case")]
#[non_exhaustive]
pub enum CallResult {
Ok {
id: u64,
value: CallValue,
},
Denied {
id: u64,
reason: DenialReason,
detail: String,
},
}
pub const MAX_DETAIL_CHARS: usize = 512;
impl CallResult {
pub fn denied(id: u64, reason: DenialReason, detail: impl Into<String>) -> Self {
let detail = detail.into();
let mut chars = detail.chars();
let kept: String = chars.by_ref().take(MAX_DETAIL_CHARS).collect();
let kept = if chars.next().is_some() {
format!("{kept}…")
} else {
kept
};
Self::Denied {
id,
reason,
detail: super::manifest::rejected(&kept),
}
}
#[must_use]
pub const fn id(&self) -> u64 {
match self {
Self::Ok { id, .. } | Self::Denied { id, .. } => *id,
}
}
#[must_use]
pub const fn denial(&self) -> Option<DenialReason> {
match self {
Self::Ok { .. } => None,
Self::Denied { reason, .. } => Some(*reason),
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
#[non_exhaustive]
pub enum CallValue {
Done,
Value {
value: PluginValue,
found: bool,
},
RowId {
row_id: String,
},
Rows {
rows: Vec<PluginRow>,
truncated: bool,
},
Http {
status: u16,
headers: Vec<(String, String)>,
body: String,
},
JobId {
job_id: String,
},
}
#[derive(Clone, Default)]
pub struct CapabilityServices {
pub tenant: Option<String>,
pub kv: Option<Arc<dyn KvStore>>,
pub http: Option<Arc<dyn OutboundHttp>>,
pub db: Option<Arc<dyn PluginStore>>,
pub jobs: Option<Arc<dyn JobSink>>,
pub rate: Option<Arc<quota::CapabilityRateLimiter>>,
}
impl fmt::Debug for CapabilityServices {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("CapabilityServices")
.field("tenant", &self.tenant)
.field("kv", &self.kv.is_some())
.field("http", &self.http.is_some())
.field("db", &self.db.is_some())
.field("jobs", &self.jobs.is_some())
.field("rate", &self.rate.is_some())
.finish()
}
}
impl CapabilityServices {
#[must_use]
pub fn none() -> Self {
Self::default()
}
#[must_use]
pub fn for_tenant(mut self, tenant: impl Into<String>) -> Self {
self.tenant = Some(tenant.into());
self
}
}
pub const NO_TENANT: &str = "-";
const TENANT_PRESENT: char = 't';
#[must_use]
pub fn tenant_segment(tenant: Option<&str>) -> String {
tenant.map_or_else(
|| NO_TENANT.to_owned(),
|id| format!("{TENANT_PRESENT}{id}"),
)
}
pub struct CapabilityRuntime {
pub(super) plugin: String,
pub(super) artifact_sha256: String,
capabilities: Vec<SandboxCapability>,
pub(super) grants: super::grants::CapabilityGrants,
pub(super) services: CapabilityServices,
quotas: QuotaLedger,
rate: Option<Arc<quota::CapabilityRateLimiter>>,
events: Vec<Event>,
dropped: Option<DroppedCalls>,
warned: Vec<(&'static str, Outcome)>,
}
const MAX_WARNED: usize = 64;
fn bounded_target(target: &str) -> String {
let mut chars = target.chars();
let kept: String = chars.by_ref().take(MAX_TARGET_CHARS).collect();
if chars.next().is_some() {
format!("{kept}…")
} else {
kept
}
}
pub const MAX_TARGET_CHARS: usize = 96;
pub const MAX_EVENTS: usize = 512;
impl fmt::Debug for CapabilityRuntime {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("CapabilityRuntime")
.field("plugin", &self.plugin)
.field("capabilities", &self.capabilities)
.field("events", &self.events.len())
.finish_non_exhaustive()
}
}
impl CapabilityRuntime {
#[must_use]
pub fn new(manifest: &SandboxManifest, services: CapabilityServices) -> Self {
Self {
plugin: manifest.name.clone(),
artifact_sha256: manifest.sha256.clone(),
capabilities: manifest.capabilities.clone(),
grants: manifest.grants.clone(),
rate: services.rate.clone(),
services,
quotas: QuotaLedger::new(manifest.quotas),
events: Vec::new(),
dropped: None,
warned: Vec::new(),
}
}
#[must_use]
pub fn with_artifact_sha256(mut self, digest: impl Into<String>) -> Self {
self.artifact_sha256 = digest.into();
self
}
#[must_use]
pub const fn quotas(&self) -> &CapabilityQuotas {
self.quotas.declared()
}
#[must_use]
pub fn tenant(&self) -> Option<&str> {
self.services.tenant.as_deref()
}
#[must_use]
pub fn tenant_key(&self) -> String {
tenant_segment(self.tenant())
}
#[must_use]
pub fn events(&self) -> &[Event] {
&self.events
}
#[must_use]
pub fn take_events(&mut self) -> Vec<Event> {
std::mem::take(&mut self.events)
}
#[must_use]
pub const fn dropped_events(&self) -> u64 {
match self.dropped {
Some(dropped) => dropped.count,
None => 0,
}
}
#[must_use]
pub const fn dropped_calls(&self) -> Option<DroppedCalls> {
self.dropped
}
pub(crate) const fn reply_reserve(&self, call: &CapabilityCall) -> usize {
let quotas = self.quotas();
let reserve =
call.reply_reserve_for(quotas.kv_value_bytes as usize, quotas.db_rows as usize);
match call {
CapabilityCall::HttpFetch { .. } => {
let body = (quotas.outbound_response_bytes as usize).saturating_mul(6);
reserve.saturating_sub(MAX_RESULT_BYTES.saturating_sub(body))
}
_ => reserve,
}
}
#[must_use]
pub fn is_granted(&self, capability: SandboxCapability) -> bool {
self.capabilities.contains(&capability)
}
pub fn dispatch(&mut self, call: &CapabilityCall) -> CallResult {
self.dispatch_within(call, usize::MAX)
}
pub fn dispatch_within(&mut self, call: &CapabilityCall, reply_room: usize) -> CallResult {
self.dispatch_checked(call, reply_room, |_| true)
}
pub(crate) fn dispatch_checked(
&mut self,
call: &CapabilityCall,
reply_room: usize,
fuel_covers: impl FnOnce(usize) -> bool,
) -> CallResult {
let id = call.id();
let capability = call.capability();
let named = call.target_excerpt();
if let Err(field) = self.quotas.charge_call() {
return self.record(
call,
&named,
CallResult::denied(
id,
DenialReason::QuotaExceeded,
format!("the per-request `{field}` quota is spent"),
),
);
}
if !self.is_granted(capability) {
return self.record(
call,
&named,
CallResult::denied(
id,
DenialReason::CapabilityNotGranted,
format!("this plugin was not granted `{capability}`"),
),
);
}
let target = match self.target_of(call) {
Ok(target) => target,
Err(result) => return self.record(call, &named, result),
};
let reserve = self.reply_reserve(call);
if reply_room < reserve {
return self.record(
call,
&target,
CallResult::denied(
id,
DenialReason::QuotaExceeded,
"there is not room to deliver this call's answer; read the pending replies \
and retry",
),
);
}
if !fuel_covers(reserve) {
return self.record(
call,
&target,
CallResult::denied(
id,
DenialReason::QuotaExceeded,
"the fuel left cannot read this call's answer and finish the request; \
increase `fuel`",
),
);
}
if let Err(result) = self.validate(call) {
return self.record(call, &target, result);
}
if let Err(field) = self.quotas.charge_capability(call) {
return self.record(
call,
&target,
CallResult::denied(
id,
DenialReason::QuotaExceeded,
format!("the per-request `{field}` quota is spent"),
),
);
}
if self
.rate
.as_ref()
.is_some_and(|rate| !rate.try_take(capability))
{
return self.record(
call,
&target,
CallResult::denied(
id,
DenialReason::QuotaExceeded,
format!("this plugin is over its `{capability}` calls-per-second ceiling"),
),
);
}
let result = self.perform(call, &target);
self.record(call, &target, result)
}
fn target_of(&self, call: &CapabilityCall) -> Result<String, CallResult> {
let id = call.id();
match call {
CapabilityCall::KvGet { key, .. }
| CapabilityCall::KvSet { key, .. }
| CapabilityCall::KvDelete { key, .. } => {
if key.is_empty() || key.len() > MAX_KV_KEY_BYTES {
return Err(CallResult::denied(
id,
DenialReason::Malformed,
format!("a kv key must be 1..={MAX_KV_KEY_BYTES} bytes"),
));
}
if key.chars().any(char::is_control) {
return Err(CallResult::denied(
id,
DenialReason::Malformed,
"a kv key may not carry control characters",
));
}
Ok(key.clone())
}
CapabilityCall::HttpFetch { url, .. } => {
let host = outbound::host_of(url).ok_or_else(|| {
CallResult::denied(
id,
DenialReason::Malformed,
"expected an absolute http/https URL with a bare host and no userinfo",
)
})?;
self.require_grant(id, SandboxCapability::HttpOutbound, &host, "hosts")?;
Ok(host)
}
CapabilityCall::DbInsert { table, .. }
| CapabilityCall::DbGet { table, .. }
| CapabilityCall::DbQuery { table, .. }
| CapabilityCall::DbUpdate { table, .. }
| CapabilityCall::DbDelete { table, .. } => {
self.require_grant(id, SandboxCapability::Db, table, "tables")?;
Ok(table.clone())
}
CapabilityCall::JobEnqueue { job_type, .. } => {
self.require_grant(id, SandboxCapability::Jobs, job_type, "job_types")?;
Ok(job_type.clone())
}
}
}
fn require_grant(
&self,
id: u64,
capability: SandboxCapability,
entry: &str,
field: &'static str,
) -> Result<(), CallResult> {
if self.grants.allows(capability, entry) {
return Ok(());
}
Err(CallResult::denied(
id,
DenialReason::NotInGrant,
format!(
"`[grants].{field}` does not name {entry:?}",
entry = super::manifest::rejected(entry)
),
))
}
fn validate(&self, call: &CapabilityCall) -> Result<(), CallResult> {
match call {
CapabilityCall::KvSet { .. } => kv::validate(self, call),
CapabilityCall::KvGet { .. } | CapabilityCall::KvDelete { .. } => Ok(()),
CapabilityCall::HttpFetch { .. } => outbound::check_request(self, call).map(|_| ()),
CapabilityCall::DbInsert { .. }
| CapabilityCall::DbGet { .. }
| CapabilityCall::DbQuery { .. }
| CapabilityCall::DbUpdate { .. }
| CapabilityCall::DbDelete { .. } => db::validate(call),
CapabilityCall::JobEnqueue { .. } => jobs::validate(call),
}
}
fn perform(&self, call: &CapabilityCall, target: &str) -> CallResult {
match call {
CapabilityCall::KvGet { .. }
| CapabilityCall::KvSet { .. }
| CapabilityCall::KvDelete { .. } => kv::perform(self, call, target),
CapabilityCall::HttpFetch { .. } => outbound::perform(self, call, target),
CapabilityCall::DbInsert { .. }
| CapabilityCall::DbGet { .. }
| CapabilityCall::DbQuery { .. }
| CapabilityCall::DbUpdate { .. }
| CapabilityCall::DbDelete { .. } => db::perform(self, call, target),
CapabilityCall::JobEnqueue { .. } => jobs::perform(self, call, target),
}
}
fn record(&mut self, call: &CapabilityCall, target: &str, result: CallResult) -> CallResult {
let target = super::manifest::rejected(&bounded_target(target));
let outcome = match result.denial() {
None => Outcome::Allowed,
Some(DenialReason::QuotaExceeded) => Outcome::QuotaExceeded,
Some(reason) => Outcome::Denied(reason),
};
if matches!(outcome, Outcome::Allowed) {
tracing::debug!(
plugin = self.plugin,
capability = call.capability().as_str(),
operation = call.operation(),
target = target.as_str(),
"sandboxed plugin used a granted capability"
);
} else if self
.warned
.iter()
.any(|(operation, seen)| *operation == call.operation() && *seen == outcome)
{
} else {
tracing::warn!(
plugin = self.plugin,
capability = call.capability().as_str(),
operation = call.operation(),
target = target.as_str(),
outcome = %outcome,
"sandboxed plugin was refused a capability call"
);
if self.warned.len() < MAX_WARNED {
self.warned.push((call.operation(), outcome));
}
}
let at = crate::time::ambient_instant();
if self.events.len() < MAX_EVENTS {
self.events.push(Event {
capability: call.capability(),
operation: call.operation(),
target,
outcome,
at,
});
} else {
self.dropped = Some(self.dropped.map_or(
DroppedCalls {
count: 1,
first: at,
last: at,
},
|dropped| DroppedCalls {
count: dropped.count.saturating_add(1),
last: at,
..dropped
},
));
}
result
}
}
#[cfg(test)]
mod tests {
use super::super::grants::MAX_QUOTA;
use super::*;
use jobs::MemoryJobSink;
use kv::MemoryKvStore;
use outbound::{OutboundResponse, RecordingHttp};
fn manifest(capabilities: &[SandboxCapability]) -> SandboxManifest {
let caps = capabilities
.iter()
.map(|capability| format!("\"{}\"", capability.as_str()))
.collect::<Vec<_>>()
.join(", ");
let mut grants = String::from("[grants]\n");
if capabilities.contains(&SandboxCapability::HttpOutbound) {
grants.push_str("hosts = [\"api.example.com\"]\n");
}
if capabilities.contains(&SandboxCapability::Db) {
grants.push_str("tables = [\"orders\"]\n");
}
if capabilities.contains(&SandboxCapability::Jobs) {
grants.push_str("job_types = [\"reindex\"]\n");
}
if capabilities.contains(&SandboxCapability::Render) {
grants.push_str("slots = [\"order-summary\"]\n");
}
SandboxManifest::parse(&format!(
r#"
name = "shop"
version = "0.1.0"
wire_version = 1
prefix = "/shop"
capabilities = [{caps}]
sha256 = "{digest}"
[[routes]]
method = "GET"
path = "/shop/panel"
{grants}
"#,
digest = "c".repeat(64)
))
.expect("the fixture manifest is valid")
}
fn everything() -> Vec<SandboxCapability> {
SandboxCapability::ALL.to_vec()
}
struct Fixture {
runtime: CapabilityRuntime,
kv: Arc<MemoryKvStore>,
db: Arc<db::MemoryPluginStore>,
jobs: Arc<MemoryJobSink>,
http: Arc<RecordingHttp>,
}
fn fixture(capabilities: &[SandboxCapability], tenant: Option<&str>) -> Fixture {
let kv = MemoryKvStore::new();
let db = db::MemoryPluginStore::new();
let jobs = MemoryJobSink::new();
let http = RecordingHttp::new();
let mut services = CapabilityServices {
kv: Some(Arc::clone(&kv) as Arc<dyn KvStore>),
db: Some(Arc::clone(&db) as Arc<dyn PluginStore>),
jobs: Some(Arc::clone(&jobs) as Arc<dyn JobSink>),
http: Some(Arc::clone(&http) as Arc<dyn OutboundHttp>),
..CapabilityServices::none()
};
if let Some(tenant) = tenant {
services = services.for_tenant(tenant);
}
Fixture {
runtime: CapabilityRuntime::new(&manifest(capabilities), services),
kv,
db,
jobs,
http,
}
}
fn row(pairs: &[(&str, &str)]) -> PluginRow {
pairs
.iter()
.map(|(key, value)| ((*key).to_owned(), PluginValue::Text((*value).to_owned())))
.collect()
}
struct RogueStore {
inserted: String,
rows: Vec<PluginRow>,
}
impl db::PluginStore for RogueStore {
fn insert(&self, _: &db::Scope, _: PluginRow) -> Result<String, db::StoreError> {
Ok(self.inserted.clone())
}
fn get(&self, _: &db::Scope, _: &str) -> Result<Option<PluginRow>, db::StoreError> {
Ok(self.rows.first().cloned())
}
fn query(
&self,
_: &db::Scope,
_: &PluginRow,
_: usize,
_: usize,
_: Option<&str>,
) -> Result<db::QueryPage, db::StoreError> {
Ok(db::QueryPage {
rows: self.rows.clone(),
truncated: false,
})
}
fn update(&self, _: &db::Scope, _: &str, _: PluginRow) -> Result<(), db::StoreError> {
Ok(())
}
fn delete(&self, _: &db::Scope, _: &str) -> Result<(), db::StoreError> {
Ok(())
}
}
struct RogueSink(String);
impl JobSink for RogueSink {
fn enqueue(&self, _: jobs::PluginJob) -> Result<String, String> {
Ok(self.0.clone())
}
}
fn rogue_db(inserted: &str, rows: Vec<PluginRow>) -> CapabilityRuntime {
let services = CapabilityServices {
db: Some(Arc::new(RogueStore {
inserted: inserted.to_owned(),
rows,
}) as Arc<dyn PluginStore>),
..CapabilityServices::none()
};
CapabilityRuntime::new(&manifest(&everything()), services)
}
fn db_get() -> CapabilityCall {
CapabilityCall::DbGet {
id: 1,
table: "orders".to_owned(),
row_id: "r1".to_owned(),
}
}
#[test]
fn a_stored_row_over_the_row_ceiling_is_not_served() {
let mut fat = row(&[("row_id", "r1")]);
for column in 0..5 {
fat.insert(
format!("c{column}"),
PluginValue::Text("x".repeat(MAX_VALUE_TEXT_BYTES)),
);
}
assert!(row_weight(&fat) > MAX_ROW_BYTES && row_weight(&fat) < MAX_RESULT_BYTES);
let mut runtime = rogue_db("r1", vec![fat.clone()]);
assert_eq!(
runtime.dispatch(&db_get()).denial(),
Some(DenialReason::ResultRejected)
);
let query = CapabilityCall::DbQuery {
id: 2,
table: "orders".to_owned(),
filter: PluginRow::new(),
limit: 0,
after: None,
};
let rejected = runtime.dispatch(&query);
assert_eq!(rejected.denial(), Some(DenialReason::ResultRejected));
assert!(format!("{rejected:?}").contains("r1"), "{rejected:?}");
}
#[test]
fn a_stored_row_the_host_would_not_have_accepted_is_not_served() {
let bad_name = row(&[("row_id", "r1"), ("Bad-Name", "x")]);
let host_column = row(&[("row_id", "r1"), (db::TENANT_COLUMN, "alpha")]);
let too_wide: PluginRow = (0..=MAX_ROW_COLUMNS)
.map(|index| (format!("c{index}"), PluginValue::Int(1)))
.collect();
for stored in [bad_name, host_column, too_wide] {
let mut runtime = rogue_db("r1", vec![stored]);
assert_eq!(
runtime.dispatch(&db_get()).denial(),
Some(DenialReason::ResultRejected)
);
}
}
#[test]
fn a_stored_row_without_a_usable_id_is_not_served() {
let long = "r".repeat(MAX_ROW_ID_BYTES + 1);
for id in [
None,
Some(PluginValue::Int(7)),
Some(PluginValue::Text(String::new())),
Some(PluginValue::Text(long)),
] {
let mut stored = row(&[("sku", "A-1")]);
if let Some(id) = id {
stored.insert("row_id".to_owned(), id);
}
let mut runtime = rogue_db("r1", vec![stored]);
assert_eq!(
runtime.dispatch(&db_get()).denial(),
Some(DenialReason::ResultRejected)
);
}
}
#[test]
fn a_legal_stored_row_is_still_served_with_its_row_id() {
let mut runtime = rogue_db("r1", vec![row(&[("row_id", "r1"), ("sku", "A-1")])]);
assert!(runtime.dispatch(&db_get()).denial().is_none());
}
#[test]
fn a_store_assigned_id_the_guest_could_not_use_is_a_rejected_result() {
for bad in [String::new(), "r".repeat(MAX_ROW_ID_BYTES + 1)] {
let mut runtime = rogue_db(&bad, Vec::new());
let result = runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: row(&[("sku", "A-1")]),
});
assert_eq!(
result.denial(),
Some(DenialReason::ResultRejected),
"{} byte id",
bad.len()
);
}
}
#[test]
fn a_sink_assigned_job_id_over_the_ceiling_is_refused_but_counted_as_queued() {
let services = CapabilityServices {
jobs: Some(Arc::new(RogueSink("j".repeat(MAX_JOB_ID_BYTES + 1))) as Arc<dyn JobSink>),
..CapabilityServices::none()
};
let mut runtime = CapabilityRuntime::new(&manifest(&everything()), services);
let result = runtime.dispatch(&CapabilityCall::JobEnqueue {
id: 1,
job_type: "reindex".to_owned(),
payload: PluginRow::new(),
});
assert_eq!(result.denial(), Some(DenialReason::ResultRejected));
let log = PluginActivityLog::new();
log.ingest("shop", runtime.take_events());
let summary = log.summary("shop", std::time::Duration::from_secs(60));
assert_eq!(
summary.job_types.get("reindex").copied(),
Some(1),
"the enqueue committed: {summary:?}"
);
}
#[test]
fn a_call_refused_for_its_shape_spends_no_backend_quota() {
let mut fixture = fixture(&everything(), Some("alpha"));
let fat: PluginRow = (0..=MAX_ROW_COLUMNS)
.map(|index| (format!("c{index}"), PluginValue::Int(1)))
.collect();
for _ in 0..40 {
let result = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: fat.clone(),
});
assert_eq!(result.denial(), Some(DenialReason::Malformed));
}
let ok = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 2,
table: "orders".to_owned(),
row: row(&[("sku", "A-1")]),
});
assert!(ok.denial().is_none(), "{ok:?}");
for _ in 0..10 {
let bad = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 3,
method: "TRACE".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: Vec::new(),
body: String::new(),
});
assert_eq!(bad.denial(), Some(DenialReason::Malformed));
}
let ok = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 4,
method: "GET".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: Vec::new(),
body: String::new(),
});
assert_ne!(ok.denial(), Some(DenialReason::QuotaExceeded), "{ok:?}");
}
#[test]
fn a_huge_target_is_cut_before_it_is_copied() {
let key = "k".repeat(MAX_KV_KEY_BYTES * 1000);
let excerpt = CapabilityCall::KvGet { id: 1, key }.target_excerpt();
assert!(
excerpt.chars().count() <= MAX_TARGET_CHARS + 1,
"{}",
excerpt.len()
);
let url = format!("https://{}.test/", "h".repeat(1 << 20));
let excerpt = CapabilityCall::HttpFetch {
id: 1,
method: "GET".to_owned(),
url,
headers: Vec::new(),
body: String::new(),
}
.target_excerpt();
assert!(
excerpt.chars().count() <= MAX_TARGET_CHARS + 1,
"{}",
excerpt.len()
);
}
#[test]
fn a_row_exactly_at_the_ceiling_is_served_and_one_byte_over_is_not() {
let at_ceiling = |extra: usize| {
let mut stored = row(&[("row_id", "r1")]);
let mut left = MAX_ROW_BYTES + extra;
for column in 0..4 {
let name = format!("c{column}");
left -= name.len();
let take = left.min(MAX_VALUE_TEXT_BYTES);
left -= take;
stored.insert(name, PluginValue::Text("x".repeat(take)));
}
assert_eq!(left, 0);
stored
};
assert!(
rogue_db("r1", vec![at_ceiling(0)])
.dispatch(&db_get())
.denial()
.is_none()
);
assert_eq!(
rogue_db("r1", vec![at_ceiling(1)])
.dispatch(&db_get())
.denial(),
Some(DenialReason::ResultRejected)
);
}
#[test]
fn a_job_id_at_its_ceiling_is_returned_and_an_empty_one_is_not() {
for (id, rejected) in [("j".repeat(MAX_JOB_ID_BYTES), false), (String::new(), true)] {
let services = CapabilityServices {
jobs: Some(Arc::new(RogueSink(id)) as Arc<dyn JobSink>),
..CapabilityServices::none()
};
let mut runtime = CapabilityRuntime::new(&manifest(&everything()), services);
let result = runtime.dispatch(&CapabilityCall::JobEnqueue {
id: 1,
job_type: "reindex".to_owned(),
payload: PluginRow::new(),
});
assert_eq!(
result.denial() == Some(DenialReason::ResultRejected),
rejected,
"{result:?}"
);
}
}
#[test]
fn a_query_reserve_follows_its_page_and_fits_the_queue() {
let query = |limit| CapabilityCall::DbQuery {
id: 1,
table: "orders".to_owned(),
filter: PluginRow::new(),
limit,
after: None,
};
let huge = 1_000_000;
assert!(query(1).reply_reserve_for(0, huge) < query(0).reply_reserve_for(0, huge));
assert!(query(0).reply_reserve_for(0, huge) <= super::super::host::MAX_QUEUED_REPLY_BYTES);
assert!(query(1).reply_reserve_for(0, huge) < MAX_RESULT_BYTES + 8192);
}
#[test]
fn reply_room_is_checked_at_its_exact_edge() {
let post = CapabilityCall::HttpFetch {
id: 1,
method: "POST".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: Vec::new(),
body: String::new(),
};
let reserve = post.reply_reserve();
let mut fixture = fixture(&everything(), Some("alpha"));
let short = fixture.runtime.dispatch_within(&post, reserve - 1);
assert_eq!(short.denial(), Some(DenialReason::QuotaExceeded));
assert!(fixture.http.seen().is_empty(), "the POST ran");
let fits = fixture.runtime.dispatch_within(&post, reserve);
assert_ne!(fits.denial(), Some(DenialReason::QuotaExceeded), "{fits:?}");
let get = CapabilityCall::KvGet {
id: 2,
key: "cart".to_owned(),
};
let read = get.reply_reserve_for(fixture.runtime.quotas().kv_value_bytes as usize, 0);
assert_eq!(
fixture.runtime.dispatch_within(&get, read - 1).denial(),
Some(DenialReason::QuotaExceeded)
);
assert!(
fixture
.runtime
.dispatch_within(&get, read)
.denial()
.is_none()
);
}
#[test]
fn an_oversized_kv_value_is_refused_without_spending_the_write_quota() {
let mut fixture = fixture(&everything(), Some("alpha"));
let huge = PluginValue::Text("x".repeat(64 * 1024 + 1));
for _ in 0..40 {
let result = fixture.runtime.dispatch(&CapabilityCall::KvSet {
id: 1,
key: "cart".to_owned(),
value: huge.clone(),
});
assert_eq!(result.denial(), Some(DenialReason::QuotaExceeded));
}
let ok = fixture.runtime.dispatch(&CapabilityCall::KvSet {
id: 2,
key: "cart".to_owned(),
value: PluginValue::Int(1),
});
assert!(ok.denial().is_none(), "{ok:?}");
}
#[test]
fn an_ungranted_capability_is_denied_and_recorded() {
let mut fixture = fixture(&[SandboxCapability::HttpRequest], None);
let result = fixture.runtime.dispatch(&CapabilityCall::KvGet {
id: 1,
key: "cart".to_owned(),
});
assert_eq!(result.denial(), Some(DenialReason::CapabilityNotGranted));
assert_eq!(fixture.runtime.events().len(), 1);
assert_eq!(
fixture.runtime.events().first().map(|event| event.outcome),
Some(audit::CapabilityOutcome::Denied(
DenialReason::CapabilityNotGranted
))
);
assert!(fixture.kv.keys().is_empty(), "nothing was written");
}
#[test]
fn a_refusal_records_what_was_reached_for_not_a_dash() {
let mut granted = fixture(&everything(), Some("alpha"));
let _ = granted.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "GET".to_owned(),
url: "https://attacker.test/steal".to_owned(),
headers: Vec::new(),
body: String::new(),
});
let _ = granted.runtime.dispatch(&CapabilityCall::DbQuery {
id: 2,
table: "users".to_owned(),
filter: PluginRow::new(),
limit: 1,
after: None,
});
let _ = granted.runtime.dispatch(&CapabilityCall::JobEnqueue {
id: 3,
job_type: "drain-accounts".to_owned(),
payload: PluginRow::new(),
});
let targets: Vec<&str> = granted
.runtime
.events()
.iter()
.map(|event| event.target.as_str())
.collect();
assert_eq!(targets, vec!["attacker.test", "users", "drain-accounts"]);
let mut hostile = fixture(&[SandboxCapability::HttpRequest], None);
let _ = hostile.runtime.dispatch(&CapabilityCall::KvGet {
id: 4,
key: "a\u{1b}[2K\rforged".to_owned(),
});
let recorded = hostile
.runtime
.events()
.first()
.map(|event| event.target.clone())
.unwrap_or_default();
assert!(!recorded.contains('\u{1b}'), "{recorded:?}");
}
#[test]
fn an_ungranted_call_cannot_spend_the_shared_call_budget() {
let mut ungranted = fixture(&[SandboxCapability::HttpRequest], None);
for id in 0..1_000 {
let _ = ungranted.runtime.dispatch(&CapabilityCall::KvGet {
id,
key: "cart".to_owned(),
});
}
let mut granted = fixture(&everything(), None);
assert!(
granted
.runtime
.dispatch(&CapabilityCall::KvGet {
id: 1,
key: "cart".to_owned(),
})
.denial()
.is_none()
);
}
#[derive(Debug)]
struct UnreachableKvStore;
impl KvStore for UnreachableKvStore {
fn get(&self, _key: &str) -> Result<Option<PluginValue>, String> {
Err("the key/value backend is unreachable".to_owned())
}
fn set(&self, _key: &str, _value: PluginValue) -> Result<(), String> {
Err("the key/value backend is unreachable".to_owned())
}
fn delete(&self, _key: &str) -> Result<(), String> {
Err("the key/value backend is unreachable".to_owned())
}
}
#[test]
fn a_kv_backend_outage_reaches_the_guest_as_a_failure_not_a_miss() {
let manifest = manifest(&everything());
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
kv: Some(Arc::new(UnreachableKvStore) as Arc<dyn KvStore>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
let result = runtime.dispatch(&CapabilityCall::KvGet {
id: 1,
key: "cart".to_owned(),
});
assert!(
matches!(
result,
CallResult::Denied {
id: 1,
reason: DenialReason::BackendError,
..
}
),
"an unreachable store must deny, not report an absent key: {result:?}"
);
let result = runtime.dispatch(&CapabilityCall::KvDelete {
id: 2,
key: "cart".to_owned(),
});
assert!(
matches!(
result,
CallResult::Denied {
id: 2,
reason: DenialReason::BackendError,
..
}
),
"an unreachable store must deny the delete too: {result:?}"
);
}
#[test]
fn an_overlong_outbound_method_is_refused_without_being_copied() {
let manifest = manifest(&everything());
let http = RecordingHttp::new();
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
http: Some(Arc::clone(&http) as Arc<dyn OutboundHttp>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
let result = runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "G".repeat(1024 * 1024),
url: "https://api.example.com/orders".to_owned(),
headers: Vec::new(),
body: String::new(),
});
assert!(
matches!(
result,
CallResult::Denied {
id: 1,
reason: DenialReason::Malformed,
..
}
),
"an overlong method is not a method: {result:?}"
);
let lowercase = runtime.dispatch(&CapabilityCall::HttpFetch {
id: 2,
method: "post".to_owned(),
url: "https://api.example.com/orders".to_owned(),
headers: Vec::new(),
body: String::new(),
});
assert!(
!matches!(
lowercase,
CallResult::Denied {
reason: DenialReason::Malformed,
..
}
),
"`post` must still be POST: {lowercase:?}"
);
}
#[test]
fn one_tenants_key_is_unreadable_when_another_is_active() {
let store = MemoryKvStore::new();
let services = |tenant: &str| {
CapabilityServices {
kv: Some(Arc::clone(&store) as Arc<dyn KvStore>),
..CapabilityServices::none()
}
.for_tenant(tenant)
};
let manifest = manifest(&everything());
let mut alpha = CapabilityRuntime::new(&manifest, services("alpha"));
let _ = alpha.dispatch(&CapabilityCall::KvSet {
id: 1,
key: "cart".to_owned(),
value: PluginValue::Text("alpha's".to_owned()),
});
let mut beta = CapabilityRuntime::new(&manifest, services("beta"));
let result = beta.dispatch(&CapabilityCall::KvGet {
id: 2,
key: "cart".to_owned(),
});
assert_eq!(
result,
CallResult::Ok {
id: 2,
value: CallValue::Value {
value: PluginValue::Null,
found: false
}
},
"tenant beta must not see tenant alpha's key"
);
}
#[test]
fn no_tenant_key_can_be_spelled_to_reach_another_tenants_namespace() {
let store = MemoryKvStore::new();
let manifest = manifest(&everything());
let services = |tenant: &str| {
CapabilityServices {
kv: Some(Arc::clone(&store) as Arc<dyn KvStore>),
..CapabilityServices::none()
}
.for_tenant(tenant)
};
let mut beta = CapabilityRuntime::new(&manifest, services("b"));
let _ = beta.dispatch(&CapabilityCall::KvSet {
id: 1,
key: "cart".to_owned(),
value: PluginValue::Text("beta's".to_owned()),
});
let mut alpha = CapabilityRuntime::new(&manifest, services("a"));
let result = alpha.dispatch(&CapabilityCall::KvGet {
id: 2,
key: "b:cart".to_owned(),
});
assert_eq!(
result,
CallResult::Ok {
id: 2,
value: CallValue::Value {
value: PluginValue::Null,
found: false
}
}
);
assert_eq!(store.keys().len(), 1, "{:?}", store.keys());
}
#[test]
fn a_tenant_id_carrying_the_separator_cannot_collide_with_another() {
assert_ne!(
kv::namespaced_key("shop", Some("b%3Ax"), "k"),
kv::namespaced_key("shop", Some("b"), "%3Ax:k")
);
assert_ne!(
kv::namespaced_key("shop", Some("a"), "b:c"),
kv::namespaced_key("shop", Some("a:b"), "c")
);
}
#[test]
fn two_plugins_never_share_a_key() {
assert_ne!(
kv::namespaced_key("shop", Some("t"), "cart"),
kv::namespaced_key("other", Some("t"), "cart")
);
}
#[test]
fn no_tenant_is_not_a_tenant_that_could_be_named() {
for spelling in [NO_TENANT, "_", "", "-", "t", "t-"] {
assert_ne!(
tenant_segment(Some(spelling)),
tenant_segment(None),
"a tenant named {spelling:?} reached the single-tenant namespace"
);
assert_ne!(
kv::namespaced_key("shop", Some(spelling), "cart"),
kv::namespaced_key("shop", None, "cart"),
"{spelling:?}"
);
}
assert_ne!(tenant_segment(Some("a")), tenant_segment(Some("b")));
}
#[test]
fn a_stored_value_over_the_current_ceiling_is_refused_on_the_way_out() {
let kv = MemoryKvStore::new();
let manifest = manifest(&everything());
let key = kv::namespaced_key(&manifest.name, Some("alpha"), "cart");
let _ = kv.set(&key, PluginValue::Text("x".repeat(4096)));
let mut quotas = manifest.quotas;
quotas.kv_value_bytes = 64;
let tightened = SandboxManifest { quotas, ..manifest };
let mut runtime = CapabilityRuntime::new(
&tightened,
CapabilityServices {
kv: Some(Arc::clone(&kv) as Arc<dyn KvStore>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
let result = runtime.dispatch(&CapabilityCall::KvGet {
id: 1,
key: "cart".to_owned(),
});
assert_eq!(
result.denial(),
Some(DenialReason::QuotaExceeded),
"{result:?}"
);
}
#[test]
fn a_query_filter_may_not_carry_the_row_id_that_would_widen_it() {
let mut fixture = fixture(&everything(), Some("alpha"));
for sku in ["A-1", "A-2", "A-3"] {
let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: row(&[("sku", sku)]),
});
assert!(inserted.denial().is_none(), "{inserted:?}");
}
let result = fixture.runtime.dispatch(&CapabilityCall::DbQuery {
id: 2,
table: "orders".to_owned(),
filter: row(&[(db::ID_COLUMN, "r1")]),
limit: 0,
after: None,
});
assert_eq!(result.denial(), Some(DenialReason::Malformed), "{result:?}");
}
#[test]
fn a_truncated_page_can_be_continued_from_where_it_stopped() {
let mut fixture = fixture(&everything(), Some("alpha"));
for sku in ["A-1", "A-2", "A-3", "A-4"] {
let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: row(&[("sku", sku)]),
});
assert!(inserted.denial().is_none(), "{inserted:?}");
}
let page = |runtime: &mut CapabilityRuntime, after: Option<String>| {
let result = runtime.dispatch(&CapabilityCall::DbQuery {
id: 2,
table: "orders".to_owned(),
filter: PluginRow::new(),
limit: 2,
after,
});
let CallResult::Ok {
value: CallValue::Rows { rows, .. },
..
} = result
else {
panic!("the query should have succeeded: {result:?}")
};
rows
};
let first = page(&mut fixture.runtime, None);
assert_eq!(first.len(), 2, "{first:?}");
let last = first
.last()
.and_then(|row| row.get(db::ID_COLUMN))
.map(ToString::to_string)
.expect("every returned row carries its id");
let second = page(&mut fixture.runtime, Some(last));
assert_eq!(second.len(), 2, "{second:?}");
for row in &second {
assert!(!first.contains(row), "{row:?} came back twice");
}
}
#[test]
fn a_row_at_the_column_ceiling_survives_a_read_modify_write() {
let mut fixture = fixture(&everything(), Some("alpha"));
let full: PluginRow = (0..MAX_ROW_COLUMNS)
.map(|index| {
(
format!("c{index}"),
PluginValue::Int(i64::try_from(index).unwrap_or(0)),
)
})
.collect();
assert_eq!(full.len(), MAX_ROW_COLUMNS);
let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: full,
});
let CallResult::Ok {
value: CallValue::RowId { row_id },
..
} = inserted
else {
panic!("a row at the ceiling should store: {inserted:?}")
};
let read = fixture.runtime.dispatch(&CapabilityCall::DbGet {
id: 2,
table: "orders".to_owned(),
row_id: row_id.clone(),
});
let CallResult::Ok {
value: CallValue::Rows { rows, .. },
..
} = read
else {
panic!("the read should have succeeded: {read:?}")
};
let mut echoed = rows.first().cloned().unwrap_or_default();
assert_eq!(
echoed.len(),
MAX_ROW_COLUMNS + 1,
"the row comes back with the host's id, which is the whole problem"
);
echoed.insert("c0".to_owned(), PluginValue::Int(99));
let updated = fixture.runtime.dispatch(&CapabilityCall::DbUpdate {
id: 3,
table: "orders".to_owned(),
row_id,
row: echoed,
});
assert!(
updated.denial().is_none(),
"writing back what was read must not be refused: {updated:?}"
);
}
#[test]
fn a_page_cut_by_the_row_quota_says_so_too() {
let base = manifest(&everything());
let mut quotas = base.quotas;
quotas.db_rows = 2;
let manifest = SandboxManifest { quotas, ..base };
let store = db::MemoryPluginStore::new();
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
db: Some(Arc::clone(&store) as Arc<dyn PluginStore>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
for sku in ["A-1", "A-2", "A-3", "A-4"] {
let inserted = runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: row(&[("sku", sku)]),
});
assert!(inserted.denial().is_none(), "{inserted:?}");
}
let result = runtime.dispatch(&CapabilityCall::DbQuery {
id: 2,
table: "orders".to_owned(),
filter: PluginRow::new(),
limit: 0,
after: None,
});
let CallResult::Ok {
value: CallValue::Rows { rows, truncated },
..
} = result
else {
panic!("the query should have succeeded: {result:?}")
};
assert_eq!(rows.len(), 2, "the quota caps the page");
assert!(truncated, "and the guest is told there is more");
}
#[test]
fn a_read_modify_write_echo_still_parses_at_the_column_ceiling() {
let columns = (0..MAX_ROW_COLUMNS)
.map(|index| format!(r#""c{index}":"v""#))
.chain(std::iter::once(r#""row_id":"r1""#.to_owned()))
.collect::<Vec<_>>()
.join(",");
let line = format!(
r#"{{"call":"db-update","id":1,"table":"orders","row_id":"r1","row":{{{columns}}}}}"#
);
let parsed = serde_json::from_str::<CapabilityCall>(&line);
assert!(
parsed.is_ok(),
"the echo of a full row must survive parsing: {parsed:?}"
);
let columns = (0..=MAX_ROW_COLUMNS)
.map(|index| format!(r#""c{index}":"v""#))
.chain(std::iter::once(r#""row_id":"r1""#.to_owned()))
.collect::<Vec<_>>()
.join(",");
let line = format!(
r#"{{"call":"db-update","id":1,"table":"orders","row_id":"r1","row":{{{columns}}}}}"#
);
assert!(
serde_json::from_str::<CapabilityCall>(&line).is_err(),
"a row genuinely over the ceiling is still refused"
);
}
#[test]
fn the_root_of_a_fragment_is_bounded_like_its_branches() {
let nodes = (0..=render::MAX_NODES)
.map(|_| r#"{"node":"text","text":""}"#)
.collect::<Vec<_>>()
.join(",");
let line = format!(r#"{{"op":"fragment","nodes":[{nodes}]}}"#);
let err = serde_json::from_str::<crate::plugin_sandbox::wire::GuestFrame>(&line)
.expect_err("a fragment over the node ceiling must not parse");
assert!(err.to_string().contains("more than"), "{err}");
let nodes = (0..render::MAX_NODES)
.map(|_| r#"{"node":"text","text":""}"#)
.collect::<Vec<_>>()
.join(",");
let line = format!(r#"{{"op":"fragment","nodes":[{nodes}]}}"#);
assert!(
serde_json::from_str::<crate::plugin_sandbox::wire::GuestFrame>(&line).is_ok(),
"a fragment at the ceiling is legal"
);
}
#[test]
fn a_guest_collection_is_refused_while_it_is_read_not_after() {
let pairs = (0..=MAX_OUTBOUND_HEADERS)
.map(|_| r#"["",""]"#)
.collect::<Vec<_>>()
.join(",");
let line = format!(
r#"{{"call":"http-fetch","id":1,"method":"GET","url":"https://api.example.com/","headers":[{pairs}],"body":""}}"#
);
let parsed = serde_json::from_str::<CapabilityCall>(&line);
let err = parsed.expect_err("a frame over the header ceiling must not parse");
assert!(
err.to_string().contains("more than"),
"the parser should say why: {err}"
);
let pairs = (0..MAX_OUTBOUND_HEADERS)
.map(|_| r#"["a","b"]"#)
.collect::<Vec<_>>()
.join(",");
let line = format!(
r#"{{"call":"http-fetch","id":1,"method":"GET","url":"https://api.example.com/","headers":[{pairs}],"body":""}}"#
);
assert!(
serde_json::from_str::<CapabilityCall>(&line).is_ok(),
"a frame at the ceiling is legal"
);
}
#[test]
fn a_declared_host_is_reachable_through_the_framework_client() {
let mut fixture = fixture(&everything(), Some("alpha"));
fixture.http.answer(
"https://api.example.com/v1/orders",
OutboundResponse {
status: 200,
headers: vec![("content-type".to_owned(), "application/json".to_owned())],
body: "[]".to_owned(),
final_url: "https://api.example.com/v1/orders".to_owned(),
},
);
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "get".to_owned(),
url: "https://api.example.com/v1/orders".to_owned(),
headers: vec![("accept".to_owned(), "application/json".to_owned())],
body: String::new(),
});
assert_eq!(
result,
CallResult::Ok {
id: 1,
value: CallValue::Http {
status: 200,
headers: vec![("content-type".to_owned(), "application/json".to_owned())],
body: "[]".to_owned(),
}
}
);
assert_eq!(fixture.http.seen().len(), 1);
}
#[test]
fn every_shape_of_undeclared_host_is_denied_and_never_leaves() {
let mut fixture = fixture(&everything(), Some("alpha"));
for url in [
"https://attacker.test/",
"https://api.example.com.attacker.test/",
"https://evil-api.example.com/",
"https://api.example.com@attacker.test/",
"https://attacker.test/?to=api.example.com",
"http://attacker.test/api.example.com",
"file:///etc/passwd",
"//api.example.com/",
"/relative",
"https://API.EXAMPLE.COM.attacker.test/",
"https://[::1]/",
] {
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 9,
method: "GET".to_owned(),
url: url.to_owned(),
headers: Vec::new(),
body: String::new(),
});
assert!(
matches!(
result.denial(),
Some(DenialReason::NotInGrant | DenialReason::Malformed)
),
"{url} was allowed: {result:?}"
);
}
assert!(
fixture.http.seen().is_empty(),
"not one of those may have left: {:?}",
fixture.http.seen()
);
}
#[test]
fn a_redirect_off_the_granted_host_is_refused_after_the_fact() {
let mut fixture = fixture(&everything(), Some("alpha"));
fixture
.http
.answer_from("https://api.example.com/r", "https://attacker.test/collect");
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "POST".to_owned(),
url: "https://api.example.com/r".to_owned(),
headers: Vec::new(),
body: "tenant data".to_owned(),
});
assert_eq!(
result.denial(),
Some(DenialReason::NotInGrant),
"{result:?}"
);
let events = fixture.runtime.events();
assert!(
events
.iter()
.any(|event| event.outcome != audit::CapabilityOutcome::Allowed),
"{events:?}"
);
}
#[test]
fn an_implementation_is_handed_what_it_needs_to_refuse_a_hop_itself() {
let mut fixture = fixture(&everything(), Some("alpha"));
fixture.http.answer(
"https://api.example.com/v1",
OutboundResponse::from_url("https://api.example.com/v1", 200, "ok"),
);
let _ = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "GET".to_owned(),
url: "https://api.example.com/v1".to_owned(),
headers: Vec::new(),
body: String::new(),
});
let seen = fixture.http.seen();
let sent = seen.first().expect("the call left");
assert!(
!sent.follow_redirects,
"redirects are never followed for a plugin"
);
assert_eq!(sent.allowed_hosts, vec!["api.example.com".to_owned()]);
assert!(
sent.timeout > std::time::Duration::ZERO,
"a call has a deadline"
);
}
#[test]
fn an_upstream_that_answered_counts_as_a_host_that_was_called() {
let mut manifest = manifest(&everything());
manifest.quotas.outbound_response_bytes = 4;
let http = RecordingHttp::new();
http.answer(
"https://api.example.com/big",
OutboundResponse::from_url("https://api.example.com/big", 200, "far too long"),
);
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
http: Some(Arc::clone(&http) as Arc<dyn OutboundHttp>),
..CapabilityServices::none()
},
);
let result = runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "GET".to_owned(),
url: "https://api.example.com/big".to_owned(),
headers: Vec::new(),
body: String::new(),
});
assert_eq!(result.denial(), Some(DenialReason::ResponseTooLarge));
let log = audit::PluginActivityLog::new();
log.ingest("shop", runtime.take_events());
let summary = log.summary("shop", std::time::Duration::from_secs(3600));
assert_eq!(summary.hosts.get("api.example.com"), Some(&1));
assert!(summary.refused_targets.is_empty(), "{summary:?}");
}
#[test]
fn the_same_host_on_another_port_is_the_granted_host() {
assert_eq!(
outbound::host_of("https://api.example.com:8443/v1"),
Some("api.example.com".to_owned())
);
assert_eq!(
outbound::host_of("https://api.example.com:not-a-port/"),
None
);
}
#[test]
fn a_header_a_plugin_may_not_set_is_denied_before_the_call_leaves() {
for (name, value) in [
("authorization", "Bearer stolen"),
("cookie", "session=stolen"),
("host", "attacker.test"),
("x-forwarded-for", "10.0.0.1"),
] {
let mut one = fixture(&everything(), Some("alpha"));
let result = one.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "GET".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: vec![(name.to_owned(), value.to_owned())],
body: String::new(),
});
assert_eq!(result.denial(), Some(DenialReason::NotInGrant), "{name}");
assert!(one.http.seen().is_empty(), "{name} left the host");
}
let mut fixture = fixture(&everything(), Some("alpha"));
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 2,
method: "GET".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: vec![("accept".to_owned(), "a\r\nx-evil: 1".to_owned())],
body: String::new(),
});
assert_eq!(result.denial(), Some(DenialReason::Malformed));
assert!(fixture.http.seen().is_empty());
}
#[test]
fn a_row_no_reply_could_carry_is_refused_where_it_goes_in() {
let mut fixture = fixture(&everything(), Some("alpha"));
let fat: PluginRow = (0..8)
.map(|index| {
(
format!("c{index}"),
PluginValue::Text("x".repeat(MAX_VALUE_TEXT_BYTES)),
)
})
.collect();
assert!(
row_weight(&fat) > MAX_ROW_BYTES,
"the fixture is fat enough"
);
let result = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: fat,
});
assert_eq!(result.denial(), Some(DenialReason::Malformed), "{result:?}");
assert!(fixture.db.keys().is_empty(), "{:?}", fixture.db.keys());
}
#[test]
fn a_query_is_bounded_by_bytes_and_says_when_it_was_cut() {
let mut fixture = fixture(&everything(), Some("alpha"));
let wide = "x".repeat(MAX_VALUE_TEXT_BYTES);
for index in 0..16 {
let mut row = row(&[("kind", "wide")]);
for column in 0..3 {
row.insert(format!("c{column}"), PluginValue::Text(wide.clone()));
}
row.insert("n".to_owned(), PluginValue::Int(index));
let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row,
});
assert!(inserted.denial().is_none(), "row {index}: {inserted:?}");
}
let result = fixture.runtime.dispatch(&CapabilityCall::DbQuery {
id: 2,
table: "orders".to_owned(),
filter: row(&[("kind", "wide")]),
limit: 0,
after: None,
});
let CallResult::Ok {
value: CallValue::Rows { rows, truncated },
..
} = result
else {
panic!("the query should have succeeded: {result:?}")
};
assert!(truncated, "16 wide rows do not fit one reply");
assert!(!rows.is_empty(), "and the answer is not empty either");
let carried: usize = rows.iter().map(row_weight).sum();
assert!(
carried <= MAX_RESULT_BYTES.saturating_add(MAX_ROW_BYTES),
"the answer carried {carried} bytes"
);
}
#[test]
fn an_upstream_cannot_answer_with_unbounded_headers() {
let mut fixture = fixture(&everything(), Some("alpha"));
let mut answer = OutboundResponse::from_url("https://api.example.com/", 200, "ok");
answer.headers = (0..64)
.map(|_| ("etag".to_owned(), "x".repeat(4096)))
.collect();
fixture.http.answer("https://api.example.com/", answer);
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "GET".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: Vec::new(),
body: String::new(),
});
let CallResult::Ok {
value: CallValue::Http { headers, .. },
..
} = result
else {
panic!("the fetch should have succeeded: {result:?}")
};
assert!(headers.len() <= MAX_OUTBOUND_HEADERS, "{}", headers.len());
let carried: usize = headers
.iter()
.map(|(name, value)| name.len().saturating_add(value.len()))
.sum();
assert!(
carried <= MAX_RESPONSE_HEADER_BYTES,
"the reply carried {carried} bytes of headers"
);
}
#[test]
fn an_outbound_request_body_is_bounded_like_the_response_is() {
let mut fixture = fixture(&everything(), Some("alpha"));
let ceiling = fixture.runtime.quotas().outbound_request_bytes as usize;
fixture.http.answer(
"https://api.example.com/",
OutboundResponse::from_url("https://api.example.com/", 200, "ok"),
);
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "POST".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: Vec::new(),
body: "x".repeat(ceiling + 1),
});
assert_eq!(
result.denial(),
Some(DenialReason::QuotaExceeded),
"{result:?}"
);
assert!(
fixture.http.seen().is_empty(),
"the call still left the host"
);
}
#[test]
fn an_outbound_requests_headers_count_against_its_byte_quota() {
let mut fixture = fixture(&everything(), Some("alpha"));
let ceiling = fixture.runtime.quotas().outbound_request_bytes as usize;
fixture.http.answer(
"https://api.example.com/",
OutboundResponse::from_url("https://api.example.com/", 200, "ok"),
);
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "POST".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: vec![("user-agent".to_owned(), "x".repeat(ceiling + 1))],
body: String::new(),
});
assert_eq!(
result.denial(),
Some(DenialReason::QuotaExceeded),
"an empty body does not make an oversized request small: {result:?}"
);
assert!(
fixture.http.seen().is_empty(),
"the call still left the host"
);
}
#[test]
fn the_in_memory_kv_store_is_bounded_by_bytes_not_only_by_keys() {
let store = MemoryKvStore::with_capacities(10_000, 4096);
let mut refused = false;
for index in 0..64 {
if store
.set(&format!("k{index}"), PluginValue::Text("x".repeat(512)))
.is_err()
{
refused = true;
break;
}
}
assert!(refused, "the store grew past its byte ceiling");
assert!(
store.keys().len() < 10_000,
"and it refused long before the key ceiling"
);
let small = MemoryKvStore::with_capacities(10_000, 4096);
assert!(small.set("k", PluginValue::Text("x".repeat(64))).is_ok());
assert!(
small.set("k", PluginValue::Text("x".repeat(8192))).is_err(),
"overwriting one key with an oversized value adds no entry and still grows the store"
);
}
#[test]
fn every_in_memory_backend_is_bounded_by_bytes_not_only_by_count() {
let scope = db::Scope {
table: "plugin_shop__orders".to_owned(),
tenant: tenant_segment(Some("alpha")),
};
let store = db::MemoryPluginStore::with_capacities(10_000, 8192);
let fat = row(&[("blob", "x".repeat(2048).as_str())]);
let mut refused = false;
for _ in 0..64 {
if store.insert(&scope, fat.clone()).is_err() {
refused = true;
break;
}
}
assert!(refused, "the row store grew past its byte ceiling");
assert!(
store.keys().len() < 10_000,
"and refused far below its row ceiling"
);
let roomy = db::MemoryPluginStore::with_capacities(10_000, 8192);
let small = roomy
.insert(&scope, row(&[("blob", "x")]))
.expect("a small row fits");
assert!(
roomy
.update(&scope, &small, row(&[("blob", "x".repeat(65536).as_str())]))
.is_err(),
"replacing a small row with an oversized one adds no row and still grows the store"
);
let kv = MemoryKvStore::with_capacities(10_000, 4096);
assert!(kv.set("k", PluginValue::Text("x".repeat(64))).is_ok());
assert!(
kv.set("k", PluginValue::Text("x".repeat(8192))).is_err(),
"the same must hold for the key/value store"
);
let sink = MemoryJobSink::with_capacities(1024, 8192);
let mut refused = false;
for _ in 0..64 {
let job = PluginJob {
plugin: "shop".to_owned(),
artifact_sha256: "c".repeat(64),
tenant: Some("alpha".to_owned()),
job_type: "reindex".to_owned(),
payload: row(&[("blob", "x".repeat(2048).as_str())]),
};
if sink.enqueue(job).is_err() {
refused = true;
break;
}
}
assert!(refused, "the job queue grew past its byte ceiling");
assert!(
sink.queued().len() < 1024,
"and refused far below its depth ceiling"
);
}
#[test]
fn a_plugin_reads_and_writes_only_its_own_tenant_scoped_table() {
let mut fixture = fixture(&everything(), Some("alpha"));
let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: row(&[("sku", "A-1")]),
});
let CallResult::Ok {
value: CallValue::RowId { row_id },
..
} = inserted
else {
panic!("the insert should have succeeded: {inserted:?}")
};
let orders = db::physical_table("shop", "orders").expect("derivable");
assert_eq!(
fixture.db.keys(),
vec![(orders, tenant_segment(Some("alpha")), row_id.clone())]
);
let read = fixture.runtime.dispatch(&CapabilityCall::DbGet {
id: 2,
table: "orders".to_owned(),
row_id,
});
let CallResult::Ok {
value: CallValue::Rows { rows, .. },
..
} = read
else {
panic!("the read should have succeeded: {read:?}")
};
assert_eq!(rows.len(), 1);
}
#[test]
fn an_undeclared_table_is_denied_however_it_is_spelled() {
let mut fixture = fixture(&everything(), Some("alpha"));
for table in [
"users",
"orders_secret",
"public.users",
"orders; drop table users",
"\"users\"",
"Orders",
"plugin_shop_orders",
"",
] {
let result = fixture.runtime.dispatch(&CapabilityCall::DbQuery {
id: 1,
table: table.to_owned(),
filter: PluginRow::new(),
limit: 10,
after: None,
});
assert_eq!(
result.denial(),
Some(DenialReason::NotInGrant),
"{table} was allowed"
);
}
}
#[test]
fn the_node_ceiling_counts_the_whole_tree_not_each_vector() {
let leaves = std::iter::repeat_n(r#"{"node":"text","text":"x"}"#, 8)
.collect::<Vec<_>>()
.join(",");
let branch = format!(r#"{{"node":"element","tag":"p","children":[{leaves}]}}"#);
let wide = std::iter::repeat_n(branch.as_str(), 128)
.collect::<Vec<_>>()
.join(",");
let line = format!(r#"{{"op":"fragment","nodes":[{wide}]}}"#);
let parsed = crate::plugin_sandbox::wire::from_line::<
crate::plugin_sandbox::wire::GuestFrame,
>(&line);
assert!(
parsed.is_err(),
"a tree past the node ceiling must be refused while it is read, not after"
);
let small = std::iter::repeat_n(branch.as_str(), 4)
.collect::<Vec<_>>()
.join(",");
let line = format!(r#"{{"op":"fragment","nodes":[{small}]}}"#);
assert!(
crate::plugin_sandbox::wire::from_line::<crate::plugin_sandbox::wire::GuestFrame>(
&line
)
.is_ok(),
"a fragment inside the ceiling must still parse"
);
assert!(
crate::plugin_sandbox::wire::from_line::<crate::plugin_sandbox::wire::GuestFrame>(
&line
)
.is_ok(),
"the budget must be restored after each parse"
);
}
#[test]
fn a_page_of_rows_encodes_within_the_result_ceiling() {
let row: PluginRow = (0..MAX_ROW_COLUMNS)
.map(|i| (format!("c{i:02}"), PluginValue::Text("v".to_owned())))
.collect();
let rows = vec![row; 20_000];
let (kept, truncated) = bounded_rows(rows);
assert!(truncated);
let encoded = serde_json::to_vec(&kept).expect("encodes").len();
assert!(
encoded <= MAX_RESULT_BYTES,
"{} rows encode to {encoded} bytes",
kept.len()
);
let row: PluginRow = (0..MAX_ROW_COLUMNS)
.map(|i| (format!("c{i:02}"), PluginValue::Text("\u{1}\"".to_owned())))
.collect();
let (kept, _) = bounded_rows(vec![row; 20_000]);
let encoded = serde_json::to_vec(&kept).expect("encodes").len();
assert!(
encoded <= MAX_RESULT_BYTES,
"escaped rows encode to {encoded}"
);
}
#[test]
fn a_scalar_weighs_what_it_encodes_to() {
assert_eq!(PluginValue::Int(i64::MIN).weight(), 20);
assert_eq!(PluginValue::Int(i64::MAX).weight(), 19);
assert_eq!(PluginValue::Int(0).weight(), 1);
assert_eq!(PluginValue::Int(-1).weight(), 2);
assert_eq!(PluginValue::Int(999).weight(), 3);
assert_eq!(PluginValue::Int(-1000).weight(), 5);
assert_eq!(PluginValue::Null.weight(), 4);
assert_eq!(PluginValue::Bool(true).weight(), 4);
assert_eq!(PluginValue::Bool(false).weight(), 5);
for value in [
PluginValue::Null,
PluginValue::Bool(true),
PluginValue::Bool(false),
PluginValue::Int(0),
PluginValue::Int(-1),
PluginValue::Int(i64::MIN),
PluginValue::Int(i64::MAX),
PluginValue::Float(-2.225_073_858_507_201_4e-308),
PluginValue::Float(1.0),
] {
let encoded = serde_json::to_string(&value).expect("encodes");
assert!(
value.weight() >= encoded.len(),
"{value:?} weighs {} but encodes to {} bytes",
value.weight(),
encoded.len()
);
}
}
#[test]
fn an_oversized_first_row_is_refused_by_the_store_too() {
let store = db::MemoryPluginStore::new();
let orders = db::physical_table("shop", "orders").expect("derivable");
let tenant = tenant_segment(Some("alpha"));
let huge = "x".repeat(super::MAX_RESULT_BYTES + 1);
store.seed(&orders, &tenant, "r1", row(&[("blob", huge.as_str())]));
let page = store
.query(
&db::Scope {
table: orders,
tenant,
},
&PluginRow::new(),
10,
super::MAX_RESULT_BYTES,
None,
)
.expect("the store answers");
assert!(
page.rows.is_empty(),
"a row over the whole result budget must not be cloned out of the store"
);
assert!(
page.truncated,
"and the caller must be told there was something it did not get"
);
}
#[test]
fn a_generated_row_id_never_lands_on_a_seeded_one() {
let store = db::MemoryPluginStore::new();
let orders = db::physical_table("shop", "orders").expect("derivable");
let tenant = tenant_segment(Some("alpha"));
store.seed(&orders, &tenant, "r1", row(&[("sku", "seeded")]));
let manifest = manifest(&everything());
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
db: Some(Arc::clone(&store) as Arc<dyn PluginStore>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
let inserted = runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: row(&[("sku", "inserted")]),
});
assert!(
matches!(inserted, CallResult::Ok { .. }),
"the insert must succeed: {inserted:?}"
);
let keys = store.keys();
assert_eq!(
keys.len(),
2,
"the seeded row must survive an insert beside it: {keys:?}"
);
assert!(
keys.iter().any(|(_, _, id)| id == "r1"),
"the seeded id must still be occupied by its own row: {keys:?}"
);
}
#[test]
fn another_tenants_rows_are_invisible_and_unwritable() {
let store = db::MemoryPluginStore::new();
store.seed(
"users",
&tenant_segment(Some("alpha")),
"u1",
row(&[("email", "ceo@example.com")]),
);
let orders = db::physical_table("shop", "orders").expect("derivable");
store.seed(
&orders,
&tenant_segment(Some("beta")),
"r99",
row(&[("sku", "beta's")]),
);
let manifest = manifest(&everything());
let mut alpha = CapabilityRuntime::new(
&manifest,
CapabilityServices {
db: Some(Arc::clone(&store) as Arc<dyn PluginStore>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
let listed = alpha.dispatch(&CapabilityCall::DbQuery {
id: 1,
table: "orders".to_owned(),
filter: PluginRow::new(),
limit: 100,
after: None,
});
assert_eq!(
listed,
CallResult::Ok {
id: 1,
value: CallValue::Rows {
rows: Vec::new(),
truncated: false
}
},
"an unfiltered query returns this tenant's rows, which are none"
);
for call in [
CapabilityCall::DbGet {
id: 2,
table: "orders".to_owned(),
row_id: "r99".to_owned(),
},
CapabilityCall::DbUpdate {
id: 3,
table: "orders".to_owned(),
row_id: "r99".to_owned(),
row: row(&[("sku", "stolen")]),
},
CapabilityCall::DbDelete {
id: 4,
table: "orders".to_owned(),
row_id: "r99".to_owned(),
},
] {
let result = alpha.dispatch(&call);
let expected_empty = matches!(call, CapabilityCall::DbGet { .. });
if expected_empty {
assert_eq!(
result,
CallResult::Ok {
id: result.id(),
value: CallValue::Rows {
rows: Vec::new(),
truncated: false
}
},
"a read of another tenant's id must miss, not error"
);
} else {
assert_eq!(
result.denial(),
Some(DenialReason::BackendError),
"{call:?} must not touch another tenant's row"
);
}
}
assert_eq!(
store.keys(),
vec![
(orders, tenant_segment(Some("beta")), "r99".to_owned()),
(
"users".to_owned(),
tenant_segment(Some("alpha")),
"u1".to_owned()
),
]
);
}
#[test]
fn a_row_may_not_carry_the_column_that_decides_its_tenant() {
let mut fixture = fixture(&everything(), Some("alpha"));
let result = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: row(&[("tenant_id", "beta")]),
});
assert_eq!(result.denial(), Some(DenialReason::NotInGrant));
}
#[test]
fn a_row_id_echoed_back_is_stripped_rather_than_refused() {
let mut fixture = fixture(&everything(), Some("alpha"));
let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 1,
table: "orders".to_owned(),
row: row(&[("sku", "A-1"), ("row_id", "r999")]),
});
let CallResult::Ok {
value: CallValue::RowId { row_id },
..
} = inserted
else {
panic!("the insert should have succeeded: {inserted:?}")
};
assert_ne!(row_id, "r999", "the store assigns the id, never the guest");
let read = fixture.runtime.dispatch(&CapabilityCall::DbGet {
id: 2,
table: "orders".to_owned(),
row_id: row_id.clone(),
});
let CallResult::Ok {
value: CallValue::Rows { rows, .. },
..
} = read
else {
panic!("the read should have succeeded: {read:?}")
};
let stored = rows.first().cloned().unwrap_or_default();
assert_eq!(
stored.get("row_id"),
Some(&PluginValue::Text(row_id.clone())),
"the row comes back addressed by the id the store assigned"
);
let updated = fixture.runtime.dispatch(&CapabilityCall::DbUpdate {
id: 3,
table: "orders".to_owned(),
row_id,
row: stored,
});
assert!(updated.denial().is_none(), "{updated:?}");
}
#[test]
fn two_plugins_can_never_be_named_onto_one_physical_table() {
let victim = db::physical_table("shop", "orders_v2");
let attacker = db::physical_table("shop_orders", "v2");
assert!(victim.is_some() && attacker.is_some());
assert_ne!(victim, attacker, "a boundary shift must not collide");
let names = ["my-shop", "my.shop", "my_shop", "My_Shop", "MY.SHOP"];
let derived: Vec<String> = names
.iter()
.filter_map(|name| db::physical_table(name, "orders"))
.collect();
assert_eq!(derived.len(), names.len(), "{derived:?}");
let mut unique = derived.clone();
unique.sort();
unique.dedup();
assert_eq!(unique.len(), names.len(), "collision in {derived:?}");
for name in &derived {
assert!(
name.starts_with("plugin_")
&& name.bytes().all(|byte| byte.is_ascii_lowercase()
|| byte.is_ascii_digit()
|| byte == b'_'),
"{name}"
);
}
}
#[test]
fn a_truncated_ledger_says_so_rather_than_reporting_a_smaller_number() {
let mut manifest = manifest(&everything());
manifest.quotas.calls = MAX_QUOTA;
manifest.quotas.kv_reads = MAX_QUOTA;
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
..CapabilityServices::none()
},
);
for id in 0..(MAX_EVENTS as u64 + 25) {
let _ = runtime.dispatch(&CapabilityCall::KvGet {
id,
key: "k".to_owned(),
});
}
assert_eq!(runtime.events().len(), MAX_EVENTS);
assert_eq!(runtime.dropped_events(), 25);
let log = audit::PluginActivityLog::new();
let dropped = runtime.dropped_events();
log.ingest("shop", runtime.take_events());
log.ingest_dropped("shop", dropped);
let summary = log.summary("shop", std::time::Duration::from_secs(3600));
assert_eq!(summary.dropped, 25);
let rendered = summary.to_string();
assert!(rendered.contains("floor"), "{rendered}");
}
#[test]
fn a_dropped_call_keeps_its_own_time_not_the_time_of_ingest() {
use chrono::TimeZone as _;
use crate::time::{TickingClock, install_ambient};
let hour = std::time::Duration::from_secs(3600);
let clock = TickingClock::starting_at(
chrono::Utc
.with_ymd_and_hms(2020, 1, 1, 0, 0, 0)
.single()
.expect("a valid date"),
);
let _clock = install_ambient(Arc::new(clock.clone()));
let mut manifest = manifest(&everything());
manifest.quotas.calls = MAX_QUOTA;
manifest.quotas.kv_reads = MAX_QUOTA;
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
..CapabilityServices::none()
},
);
let log = audit::PluginActivityLog::new();
let mut kv_get = (0..).map(|id| CapabilityCall::KvGet {
id,
key: "k".to_owned(),
});
let start = crate::time::ambient_instant();
for call in kv_get.by_ref().take(MAX_EVENTS + 20) {
let _ = runtime.dispatch(&call);
}
clock.advance(hour / 2);
let later = crate::time::ambient_instant();
for call in kv_get.take(5) {
let _ = runtime.dispatch(&call);
}
clock.advance(3 * hour / 2);
let dropped = runtime.dropped_calls().expect("25 calls were dropped");
assert_eq!(dropped.count, 25);
assert_eq!(dropped.first, start);
assert_eq!(dropped.last, later);
log.ingest("shop", runtime.take_events());
log.ingest_dropped_calls("shop", dropped);
assert_eq!(log.summary("shop", hour).dropped, 0);
assert_eq!(log.summary("shop", 7 * hour / 4).dropped, 25);
}
#[test]
fn the_outbound_reply_reserve_follows_the_response_quota() {
let post = CapabilityCall::HttpFetch {
id: 1,
method: "POST".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: Vec::new(),
body: String::new(),
};
let fixture = fixture(&everything(), Some("alpha"));
assert_eq!(fixture.runtime.reply_reserve(&post), post.reply_reserve());
let mut manifest = manifest(&everything());
manifest.quotas.outbound_response_bytes = 4096;
let small = CapabilityRuntime::new(&manifest, CapabilityServices::none());
assert_eq!(
small.reply_reserve(&post),
6 * 4096 + outbound::MAX_RESPONSE_HEADER_BYTES * 6 + 4096
);
}
#[test]
fn a_call_commits_when_it_can_change_state_outside_the_guest() {
let fetch = |method: &str| CapabilityCall::HttpFetch {
id: 1,
method: method.to_owned(),
url: "https://api.example.com/".to_owned(),
headers: Vec::new(),
body: String::new(),
};
let table = || "orders".to_owned();
let commits = [
CapabilityCall::KvSet {
id: 1,
key: "k".to_owned(),
value: PluginValue::Null,
},
CapabilityCall::KvDelete {
id: 1,
key: "k".to_owned(),
},
CapabilityCall::DbInsert {
id: 1,
table: table(),
row: PluginRow::new(),
},
CapabilityCall::DbUpdate {
id: 1,
table: table(),
row_id: "r1".to_owned(),
row: PluginRow::new(),
},
CapabilityCall::DbDelete {
id: 1,
table: table(),
row_id: "r1".to_owned(),
},
CapabilityCall::JobEnqueue {
id: 1,
job_type: "reindex".to_owned(),
payload: PluginRow::new(),
},
fetch("POST"),
fetch("PUT"),
fetch("PATCH"),
fetch("DELETE"),
fetch("BREW"),
];
for call in &commits {
assert!(call.commits(), "{call:?}");
}
let reads = [
CapabilityCall::KvGet {
id: 1,
key: "k".to_owned(),
},
CapabilityCall::DbGet {
id: 1,
table: table(),
row_id: "r1".to_owned(),
},
CapabilityCall::DbQuery {
id: 1,
table: table(),
filter: PluginRow::new(),
limit: 0,
after: None,
},
fetch("GET"),
fetch("HEAD"),
fetch("get"),
];
for call in &reads {
assert!(!call.commits(), "{call:?}");
}
}
#[test]
fn an_audit_target_is_bounded_before_it_is_escaped() {
let mut fixture = fixture(&everything(), Some("alpha"));
let _ = fixture.runtime.dispatch(&CapabilityCall::KvGet {
id: 1,
key: "\u{202e}".repeat(400),
});
let recorded = fixture
.runtime
.events()
.first()
.map(|event| event.target.clone())
.unwrap_or_default();
assert!(
recorded.len() <= MAX_TARGET_CHARS * 10 + 8,
"{} bytes: {recorded}",
recorded.len()
);
}
#[test]
fn the_in_memory_stores_refuse_rather_than_growing_without_a_ceiling() {
let kv = MemoryKvStore::with_capacity(2);
assert!(kv.set("a", PluginValue::Int(1)).is_ok());
assert!(kv.set("b", PluginValue::Int(1)).is_ok());
assert!(kv.set("c", PluginValue::Int(1)).is_err());
assert!(kv.set("a", PluginValue::Int(2)).is_ok());
let store = db::MemoryPluginStore::with_capacity(1);
let scope = db::Scope {
table: "plugin_shop__orders".to_owned(),
tenant: "alpha".to_owned(),
};
assert!(store.insert(&scope, PluginRow::new()).is_ok());
assert!(store.insert(&scope, PluginRow::new()).is_err());
}
#[test]
fn a_full_store_is_a_denial_the_guest_can_read() {
let mut runtime = CapabilityRuntime::new(
&manifest(&everything()),
CapabilityServices {
kv: Some(MemoryKvStore::with_capacity(0) as Arc<dyn KvStore>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
assert_eq!(
runtime
.dispatch(&CapabilityCall::KvSet {
id: 1,
key: "cart".to_owned(),
value: PluginValue::Int(1)
})
.denial(),
Some(DenialReason::BackendError)
);
}
#[test]
fn no_physical_table_is_derivable_from_a_name_the_grant_list_could_not_hold() {
assert_eq!(
db::physical_table("shop", "orders"),
Some("plugin_shop__orders".to_owned())
);
assert_eq!(
db::physical_table("my.shop-v2", "orders"),
Some("plugin_my_2eshop_2dv2__orders".to_owned())
);
for (plugin, table) in [
("shop", "users; drop table x"),
("shop", "\"users\""),
("shop", "Orders"),
("shop\"; --", "orders"),
("shop", ""),
("", "orders"),
("shop", &"o".repeat(64)),
(&"p".repeat(40), &"t".repeat(30)),
] {
assert_eq!(db::physical_table(plugin, table), None, "{plugin}/{table}");
}
}
#[test]
fn a_job_carries_the_enqueuing_plugin_and_tenant_not_the_guests_word_for_them() {
let mut fixture = fixture(&everything(), Some("alpha"));
let result = fixture.runtime.dispatch(&CapabilityCall::JobEnqueue {
id: 1,
job_type: "reindex".to_owned(),
payload: row(&[("since", "2026-01-01")]),
});
assert!(result.denial().is_none(), "{result:?}");
assert_eq!(
fixture.jobs.queued(),
vec![jobs::PluginJob {
plugin: "shop".to_owned(),
artifact_sha256: "c".repeat(64),
job_type: "reindex".to_owned(),
tenant: Some("alpha".to_owned()),
payload: row(&[("since", "2026-01-01")]),
}]
);
}
#[test]
fn an_undeclared_job_type_never_reaches_the_queue() {
let mut fixture = fixture(&everything(), Some("alpha"));
for job_type in ["send-invoice", "reindex-all", "REINDEX", ""] {
let result = fixture.runtime.dispatch(&CapabilityCall::JobEnqueue {
id: 1,
job_type: job_type.to_owned(),
payload: PluginRow::new(),
});
assert_eq!(
result.denial(),
Some(DenialReason::NotInGrant),
"{job_type}"
);
}
assert!(fixture.jobs.queued().is_empty());
}
#[test]
fn a_spent_quota_denies_the_call_and_nothing_else() {
let mut fixture = fixture(&everything(), Some("alpha"));
let ceiling = fixture.runtime.quotas().kv_reads;
for id in 0..ceiling {
let result = fixture.runtime.dispatch(&CapabilityCall::KvGet {
id: u64::from(id),
key: "cart".to_owned(),
});
assert!(result.denial().is_none(), "call {id}: {result:?}");
}
let over = fixture.runtime.dispatch(&CapabilityCall::KvGet {
id: 999,
key: "cart".to_owned(),
});
assert_eq!(over.denial(), Some(DenialReason::QuotaExceeded));
let job = fixture.runtime.dispatch(&CapabilityCall::JobEnqueue {
id: 1000,
job_type: "reindex".to_owned(),
payload: PluginRow::new(),
});
assert!(job.denial().is_none(), "{job:?}");
}
#[test]
fn one_plugins_spent_quota_does_not_touch_another_plugins() {
let manifest = manifest(&everything());
let store = MemoryKvStore::new();
let services = CapabilityServices {
kv: Some(Arc::clone(&store) as Arc<dyn KvStore>),
..CapabilityServices::none()
}
.for_tenant("alpha");
let mut first = CapabilityRuntime::new(&manifest, services.clone());
for id in 0..manifest.quotas.kv_reads {
let _ = first.dispatch(&CapabilityCall::KvGet {
id: u64::from(id),
key: "cart".to_owned(),
});
}
assert_eq!(
first
.dispatch(&CapabilityCall::KvGet {
id: 999,
key: "cart".to_owned()
})
.denial(),
Some(DenialReason::QuotaExceeded)
);
let mut second = CapabilityRuntime::new(&manifest, services);
assert!(
second
.dispatch(&CapabilityCall::KvGet {
id: 1,
key: "cart".to_owned()
})
.denial()
.is_none()
);
}
#[test]
fn the_shared_call_budget_bounds_the_sum_of_every_surface() {
let mut manifest = manifest(&everything());
manifest.quotas.calls = 3;
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
jobs: Some(MemoryJobSink::new() as Arc<dyn JobSink>),
..CapabilityServices::none()
},
);
for id in 0..3 {
assert!(
runtime
.dispatch(&CapabilityCall::KvGet {
id,
key: "k".to_owned()
})
.denial()
.is_none()
);
}
assert_eq!(
runtime
.dispatch(&CapabilityCall::JobEnqueue {
id: 4,
job_type: "reindex".to_owned(),
payload: PluginRow::new(),
})
.denial(),
Some(DenialReason::QuotaExceeded)
);
}
#[test]
fn a_calls_per_second_ceiling_is_shared_across_requests_and_kept_per_capability() {
let manifest = manifest(&everything());
let rate = Arc::new(quota::CapabilityRateLimiter::new(1));
let services = CapabilityServices {
kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
jobs: Some(MemoryJobSink::new() as Arc<dyn JobSink>),
rate: Some(Arc::clone(&rate)),
..CapabilityServices::none()
};
let mut first = CapabilityRuntime::new(&manifest, services.clone());
assert!(
first
.dispatch(&CapabilityCall::KvGet {
id: 1,
key: "k".to_owned()
})
.denial()
.is_none()
);
let mut second = CapabilityRuntime::new(&manifest, services);
assert_eq!(
second
.dispatch(&CapabilityCall::KvGet {
id: 3,
key: "k".to_owned()
})
.denial(),
Some(DenialReason::QuotaExceeded)
);
assert!(
second
.dispatch(&CapabilityCall::JobEnqueue {
id: 4,
job_type: "reindex".to_owned(),
payload: PluginRow::new()
})
.denial()
.is_none()
);
}
#[test]
fn an_oversized_row_or_key_is_refused_before_it_is_stored() {
let mut fixture = fixture(&everything(), Some("alpha"));
let long = "k".repeat(MAX_KV_KEY_BYTES + 1);
assert_eq!(
fixture
.runtime
.dispatch(&CapabilityCall::KvGet { id: 1, key: long })
.denial(),
Some(DenialReason::Malformed)
);
let wide: PluginRow = (0..=MAX_ROW_COLUMNS)
.map(|index| (format!("c{index}"), PluginValue::Int(1)))
.collect();
assert_eq!(
fixture
.runtime
.dispatch(&CapabilityCall::DbInsert {
id: 2,
table: "orders".to_owned(),
row: wide
})
.denial(),
Some(DenialReason::Malformed)
);
assert!(fixture.db.keys().is_empty());
}
#[test]
fn a_missing_backend_is_a_denial_rather_than_a_silent_success() {
let mut runtime =
CapabilityRuntime::new(&manifest(&everything()), CapabilityServices::none());
assert_eq!(
runtime
.dispatch(&CapabilityCall::KvSet {
id: 1,
key: "cart".to_owned(),
value: PluginValue::Int(1)
})
.denial(),
Some(DenialReason::Unavailable)
);
}
#[test]
fn a_response_header_a_plugin_may_not_see_never_reaches_it() {
let mut fixture = fixture(&everything(), Some("alpha"));
fixture.http.answer(
"https://api.example.com/v1",
OutboundResponse {
status: 200,
headers: vec![
("content-type".to_owned(), "application/json".to_owned()),
("set-cookie".to_owned(), "session=stolen".to_owned()),
("x-upstream-secret".to_owned(), "s3cret".to_owned()),
],
body: "{}".to_owned(),
final_url: "https://api.example.com/v1".to_owned(),
},
);
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "GET".to_owned(),
url: "https://api.example.com/v1".to_owned(),
headers: Vec::new(),
body: String::new(),
});
let CallResult::Ok {
value: CallValue::Http { headers, .. },
..
} = result
else {
panic!("the call should have succeeded: {result:?}")
};
assert_eq!(
headers,
vec![("content-type".to_owned(), "application/json".to_owned())]
);
}
#[test]
fn a_method_or_header_count_outside_the_allow_list_is_refused() {
for method in ["CONNECT", "TRACE", "OPTIONS", " GET", ""] {
let mut fixture = fixture(&everything(), Some("alpha"));
let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: method.to_owned(),
url: "https://api.example.com/".to_owned(),
headers: Vec::new(),
body: String::new(),
});
assert_eq!(result.denial(), Some(DenialReason::Malformed), "{method}");
assert!(fixture.http.seen().is_empty(), "{method} left the host");
}
let mut fixture = fixture(&everything(), Some("alpha"));
let many: Vec<(String, String)> = (0..=MAX_OUTBOUND_HEADERS)
.map(|_| ("accept".to_owned(), "*/*".to_owned()))
.collect();
assert_eq!(
fixture
.runtime
.dispatch(&CapabilityCall::HttpFetch {
id: 2,
method: "GET".to_owned(),
url: "https://api.example.com/".to_owned(),
headers: many,
body: String::new(),
})
.denial(),
Some(DenialReason::Malformed)
);
assert!(fixture.http.seen().is_empty());
}
#[test]
fn every_byte_ceiling_a_quota_names_is_actually_applied() {
let mut manifest = manifest(&everything());
manifest.quotas.kv_value_bytes = 8;
manifest.quotas.db_rows = 2;
manifest.quotas.outbound_response_bytes = 8;
let http = RecordingHttp::new();
http.answer(
"https://api.example.com/big",
OutboundResponse::from_url("https://api.example.com/big", 200, "x".repeat(64)),
);
let db = db::MemoryPluginStore::new();
let mut runtime = CapabilityRuntime::new(
&manifest,
CapabilityServices {
kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
db: Some(Arc::clone(&db) as Arc<dyn PluginStore>),
http: Some(Arc::clone(&http) as Arc<dyn OutboundHttp>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
assert_eq!(
runtime
.dispatch(&CapabilityCall::KvSet {
id: 1,
key: "cart".to_owned(),
value: PluginValue::Text("far too long a value".to_owned())
})
.denial(),
Some(DenialReason::QuotaExceeded)
);
for index in 0..5 {
let inserted = runtime.dispatch(&CapabilityCall::DbInsert {
id: 10 + index,
table: "orders".to_owned(),
row: row(&[("sku", "A")]),
});
assert!(inserted.denial().is_none(), "{inserted:?}");
}
for (id, limit) in [(20, 0), (21, 100)] {
let result = runtime.dispatch(&CapabilityCall::DbQuery {
id,
table: "orders".to_owned(),
filter: PluginRow::new(),
limit,
after: None,
});
let CallResult::Ok {
value: CallValue::Rows { rows, .. },
..
} = result
else {
panic!("the query should have succeeded: {result:?}")
};
assert_eq!(rows.len(), 2, "limit {limit} is clamped to `db_rows`");
}
assert_eq!(
runtime
.dispatch(&CapabilityCall::HttpFetch {
id: 30,
method: "GET".to_owned(),
url: "https://api.example.com/big".to_owned(),
headers: Vec::new(),
body: String::new(),
})
.denial(),
Some(DenialReason::ResponseTooLarge)
);
}
#[test]
fn a_row_id_outside_its_bounds_is_refused_before_the_store_sees_it() {
let mut fixture = fixture(&everything(), Some("alpha"));
for row_id in [String::new(), "r".repeat(MAX_ROW_ID_BYTES + 1)] {
assert_eq!(
fixture
.runtime
.dispatch(&CapabilityCall::DbGet {
id: 1,
table: "orders".to_owned(),
row_id: row_id.clone(),
})
.denial(),
Some(DenialReason::Malformed),
"{} bytes",
row_id.len()
);
}
}
#[test]
fn a_full_job_queue_is_a_denial_the_guest_can_read() {
let sink = jobs::MemoryJobSink::bounded(1);
let mut runtime = CapabilityRuntime::new(
&manifest(&everything()),
CapabilityServices {
jobs: Some(Arc::clone(&sink) as Arc<dyn JobSink>),
..CapabilityServices::none()
}
.for_tenant("alpha"),
);
let enqueue = |id| CapabilityCall::JobEnqueue {
id,
job_type: "reindex".to_owned(),
payload: PluginRow::new(),
};
assert!(runtime.dispatch(&enqueue(1)).denial().is_none());
assert_eq!(
runtime.dispatch(&enqueue(2)).denial(),
Some(DenialReason::BackendError)
);
assert_eq!(sink.queued().len(), 1);
}
#[test]
fn the_zero_configuration_job_queue_is_finite() {
let sink = jobs::MemoryJobSink::new();
let job = || PluginJob {
plugin: "shop".to_owned(),
artifact_sha256: "c".repeat(64),
job_type: "reindex".to_owned(),
tenant: Some("alpha".to_owned()),
payload: PluginRow::new(),
};
for index in 0..jobs::DEFAULT_JOB_DEPTH {
assert!(sink.enqueue(job()).is_ok(), "job {index}");
}
let over = sink.enqueue(job());
assert!(over.is_err(), "{over:?}");
assert_eq!(sink.queued().len(), jobs::DEFAULT_JOB_DEPTH);
}
#[test]
fn one_surface_answers_what_this_plugin_did() {
let mut fixture = fixture(&everything(), Some("alpha"));
fixture.http.answer(
"https://api.example.com/v1",
OutboundResponse {
status: 200,
headers: Vec::new(),
body: "ok".to_owned(),
final_url: "https://api.example.com/v1".to_owned(),
},
);
let _ = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 1,
method: "GET".to_owned(),
url: "https://api.example.com/v1".to_owned(),
headers: Vec::new(),
body: String::new(),
});
let _ = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
id: 2,
method: "GET".to_owned(),
url: "https://attacker.test/".to_owned(),
headers: Vec::new(),
body: String::new(),
});
let _ = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 3,
table: "orders".to_owned(),
row: row(&[("sku", "A-1")]),
});
let _ = fixture.runtime.dispatch(&CapabilityCall::JobEnqueue {
id: 4,
job_type: "reindex".to_owned(),
payload: PluginRow::new(),
});
let log = audit::PluginActivityLog::new();
log.ingest("shop", fixture.runtime.take_events());
let summary = log.summary("shop", std::time::Duration::from_secs(3600));
assert_eq!(summary.hosts.get("api.example.com"), Some(&1));
assert_eq!(summary.tables.get("plugin_shop_orders"), None);
assert_eq!(summary.tables.get("orders"), Some(&1));
assert_eq!(summary.job_types.get("reindex"), Some(&1));
assert_eq!(summary.denied.get("http-fetch"), Some(&1));
let rendered = summary.to_string();
assert!(rendered.contains("api.example.com"), "{rendered}");
assert!(rendered.contains("reindex"), "{rendered}");
}
#[test]
fn the_audit_record_never_carries_a_value() {
let mut fixture = fixture(&everything(), Some("alpha"));
let _ = fixture.runtime.dispatch(&CapabilityCall::KvSet {
id: 1,
key: "cart".to_owned(),
value: PluginValue::Text("the customer's card number".to_owned()),
});
let _ = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
id: 2,
table: "orders".to_owned(),
row: row(&[("note", "the customer's home address")]),
});
let serialized = serde_json::to_string(fixture.runtime.events()).expect("events serialize");
assert!(!serialized.contains("card number"), "{serialized}");
assert!(!serialized.contains("home address"), "{serialized}");
assert!(
serialized.contains("cart"),
"the key is a shape, not a value"
);
}
}