crates/monetize-server/src/plugins.rs
//! **The ONE box's plugins, updated from the release bucket** (owner 2026-10-17).
//!
//! ```text
//! operator ── Plugins.Update{apply} (console "Update plugins", `monetize box plugins`, the door)
//! ──▶ a QUEUED jera job (ChainQueue lane "plugins"); the call answers at once with the job
//! the job:
//! the release bucket: release/plugins/box/<box>/release-manifest.toml + .p7s
//! (`plugins_box`, fed by the kit; a config that names no box: release/plugins/release-manifest.toml)
//! ──▶ the list's own signature, against the roots this box's guest blocks trust
//! each plugin row (role plugin | vendor | cloud | dns | signup):
//! release/plugins/<blake3>/<name> + .p7s
//! ──▶ length, sha256 and blake3 as the SIGNED list names them (before a write)
//! ──▶ which guest it is (its manifest: kind + name) → the blocks that run it
//! ──▶ its own .p7s against EACH block's guest_trust_roots (before a write)
//! ──▶ the same bytes as the block's file: unchanged
//! else (apply) written beside it and RENAMED over it, .p7s first → replaced
//! the guest watcher (`crate::guests`) takes the new file in within 2 s: same gates as a start.
//! ```
//!
//! Nothing restarts. A row that does not verify is refused by name and nothing of it is
//! written; the other rows go on. The release's other artefacts (the CLI, the media receipts)
//! are not plugins and are not this verb's.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock, Weak};
use crate::config::Config;
/// Where the release's signed plugin list is in the release bucket (its signature is
/// `<key>.p7s`): written by the release build (private-monetize-ops `cargo xtask release
/// --publish`), the list LAST, so a box reading mid-publish finds the old list or the whole new one.
pub const LIST_KEY: &str = "release/plugins/release-manifest.toml";
/// **A box's own list** (OXG-5): `release/plugins/box/<box>/release-manifest.toml`, written by
/// `release --publish --box <box>` (private-monetize-ops); its signature is `<key>.p7s`.
pub fn box_list_key(box_name: &str) -> String {
format!("release/plugins/box/{box_name}/release-manifest.toml")
}
/// Pure: **the list THIS box reads**: its own box's when its config names it (`plugins_box`),
/// else the shared [`LIST_KEY`].
pub fn list_key_of(cfg: &Config) -> Result<String, String> {
match cfg.plugins_box.as_deref().map(str::trim).filter(|b| !b.is_empty()) {
None => Ok(LIST_KEY.to_string()),
Some(b) if b.bytes().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == b'-') => Ok(box_list_key(b)),
Some(b) => Err(format!("[plugins-box] plugins_box = {b:?} is not a box's name (lowercase letters, digits, hyphens): no list is read")),
}
}
/// Where one plugin of the list is: by its blake3, so a list never names bytes that a later
/// release overwrote under the same name.
pub fn object_key(blake3: &str, name: &str) -> String {
format!("release/plugins/{blake3}/{name}")
}
/// The largest list taken.
pub const MAX_LIST: usize = 256 * 1024;
/// The largest module taken.
pub const MAX_MODULE: u64 = 64 * 1024 * 1024;
/// The largest signature taken.
pub const MAX_SIGNATURE: usize = 64 * 1024;
/// The roles of the list that are plugins this verb updates.
pub const PLUGIN_ROLES: &[&str] = &["plugin", "vendor", "cloud", "dns", "signup"];
/// One row of the release's list (the shape private-monetize-ops `release::Artefact` writes).
#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize)]
pub struct Artefact {
pub name: String,
pub role: String,
pub sha256: String,
pub blake3: String,
pub bytes: u64,
}
#[derive(serde::Deserialize)]
struct List {
#[serde(default, rename = "artefact")]
artefacts: Vec<Artefact>,
}
/// Pure: the list's rows.
pub fn parse_list(text: &str) -> Result<Vec<Artefact>, String> {
toml::from_str::<List>(text).map(|l| l.artefacts).map_err(|e| format!("[plugins-list] the release's list does not parse: {e}"))
}
/// **A guest block on this box**: where its module is, which guest it must be, whose
/// signatures it trusts.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Slot {
/// `[vendors.stripe.gunnar]`
pub block: String,
/// The kind its manifest must declare (`monetize_product_guest::kinds`).
pub kind: &'static str,
/// The name its manifest must carry (a vendor instance's KIND, a product's plugin).
pub name: String,
pub module: PathBuf,
pub roots: Vec<PathBuf>,
}
fn table_files(t: &serde_json::Value) -> Option<(PathBuf, Vec<PathBuf>)> {
let module = PathBuf::from(t.get("guest")?.as_str()?);
let roots = t.get("guest_trust_roots").and_then(|v| v.as_array()).map(|a| a.iter().filter_map(|r| r.as_str().map(PathBuf::from)).collect()).unwrap_or_default();
Some((module, roots))
}
/// Pure: **every guest block of `cfg`**, the same blocks the watcher loads.
pub fn slots(cfg: &Config) -> Vec<Slot> {
use monetize_product_guest::kinds;
let mut out = Vec::new();
for (id, p) in &cfg.products {
if let Some(m) = &p.guest {
out.push(Slot { block: format!("[products.{id}]"), kind: kinds::PRODUCT, name: p.plugin.clone().unwrap_or_else(|| id.clone()), module: m.clone(), roots: p.guest_trust_roots.clone() });
}
}
for (id, v) in &cfg.vendors {
if let Some(m) = &v.guest {
let kind = id.split_once('/').map_or(id.as_str(), |(k, _)| k);
out.push(Slot { block: format!("[vendors.{}]", id.replacen('/', ".", 1)), kind: kinds::VENDOR, name: kind.to_string(), module: m.clone(), roots: v.guest_trust_roots.clone() });
}
}
let json = |t: &toml::Table| serde_json::to_value(t).unwrap_or_default();
for (name, c) in &cfg.clouds {
if let Some((module, roots)) = table_files(&json(&c.settings)) {
out.push(Slot { block: format!("[clouds.{name}]"), kind: kinds::CLOUD, name: name.clone(), module, roots });
}
}
for (name, b) in &cfg.dns_writers {
if let Some((module, roots)) = table_files(b) {
out.push(Slot { block: format!("[dns_writers.{name}]"), kind: kinds::DNS, name: name.clone(), module, roots });
}
}
for (product, b) in &cfg.signups {
if let Some((module, roots)) = table_files(&json(&b.settings)) {
out.push(Slot { block: format!("[signups.{product}]"), kind: kinds::SIGNUP, name: product.clone(), module, roots });
}
}
out
}
/// **The release bucket, read.** One object by its key, at most `cap` bytes.
pub trait Fetch: Send + Sync {
fn get(&self, object: &str, cap: usize) -> Result<Vec<u8>, String>;
}
/// What became of one plugin of the list.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Row {
pub name: String,
pub role: String,
/// `unchanged` | `replaced` | `would-replace` | `refused`
pub outcome: String,
pub detail: String,
pub blake3: String,
}
pub const UNCHANGED: &str = "unchanged";
pub const REPLACED: &str = "replaced";
pub const WOULD_REPLACE: &str = "would-replace";
pub const REFUSED: &str = "refused";
fn hex(b: &[u8]) -> String {
b.iter().map(|x| format!("{x:02x}")).collect()
}
fn sha256_hex(b: &[u8]) -> String {
use sha2::Digest as _;
hex(&sha2::Sha256::digest(b))
}
/// `bytes` with its detached signature `p7s`, verified under `roots` — in `stage`, never where
/// a block names a module. Answers the signer.
fn verify_signed(stage: &Path, file: &str, bytes: &[u8], p7s: &[u8], roots: &[PathBuf]) -> Result<String, String> {
use guest_host::ModuleTrust;
if roots.is_empty() {
return Err("no trusted roots to verify it under".into());
}
std::fs::create_dir_all(stage).map_err(|e| format!("{}: {e}", stage.display()))?;
let at = stage.join(file);
std::fs::write(ModuleTrust::signature_path(&at), p7s).map_err(|e| format!("{}: {e}", at.display()))?;
let out = ModuleTrust::required(roots.iter().cloned()).verify(&at, bytes).map_err(|e| e.to_string());
let _ = std::fs::remove_file(ModuleTrust::signature_path(&at));
out.map(|s| s.unwrap_or_default())
}
/// Write `bytes` at `to` ATOMICALLY: beside it, synced, renamed over it.
fn put(to: &Path, bytes: &[u8]) -> Result<(), String> {
use std::io::Write as _;
let dir = to.parent().ok_or_else(|| format!("{}: no directory", to.display()))?;
std::fs::create_dir_all(dir).map_err(|e| format!("{}: {e}", dir.display()))?;
let tmp = dir.join(format!(".{}.plugin-update", to.file_name().unwrap_or_default().to_string_lossy()));
let mut f = std::fs::File::create(&tmp).map_err(|e| format!("{}: {e}", tmp.display()))?;
f.write_all(bytes).and_then(|_| f.sync_all()).map_err(|e| format!("{}: {e}", tmp.display()))?;
drop(f);
std::fs::rename(&tmp, to).map_err(|e| format!("{} → {}: {e}", tmp.display(), to.display()))
}
/// **One look or update** against `cfg`'s blocks: the list fetched and verified, then each
/// plugin row (see the module docs). `Err` is a whole-job refusal (the list missing or not
/// signed); a row's refusal is its row.
pub fn update(cfg: &Config, fetch: &dyn Fetch, apply: bool, stage: &Path) -> Result<Vec<Row>, String> {
let slots = slots(cfg);
let mut roots: Vec<PathBuf> = slots.iter().flat_map(|s| s.roots.iter().cloned()).collect();
roots.sort();
roots.dedup();
if roots.is_empty() {
return Err("[plugins-no-roots] no guest block on this box names guest_trust_roots: there is nothing to verify the release's list under".into());
}
// OXG-5: this box's own list when its feed names the box; never another box's in its place.
let key = list_key_of(cfg)?;
let list = fetch.get(&key, MAX_LIST).map_err(|e| format!("[plugins-list-missing] {key}: {e}"))?;
let list_sig = fetch.get(&format!("{key}.p7s"), MAX_SIGNATURE).map_err(|e| format!("[plugins-list-unsigned] {key}.p7s: {e}"))?;
verify_signed(stage, "release-manifest.toml", &list, &list_sig, &roots).map_err(|e| format!("[plugins-list-unsigned] {key}: {e}"))?;
let text = String::from_utf8(list).map_err(|e| format!("[plugins-list] {e}"))?;
let rows = parse_list(&text)?;
let mut out = Vec::new();
for a in rows.into_iter().filter(|a| PLUGIN_ROLES.contains(&a.role.as_str())) {
let mut row = Row { name: a.name.clone(), role: a.role.clone(), outcome: REFUSED.into(), detail: String::new(), blake3: a.blake3.clone() };
match one(&a, &slots, fetch, apply, stage) {
Ok((outcome, detail)) => {
row.outcome = outcome.into();
row.detail = detail;
}
Err(why) => row.detail = why,
}
out.push(row);
}
Ok(out)
}
fn one(a: &Artefact, slots: &[Slot], fetch: &dyn Fetch, apply: bool, stage: &Path) -> Result<(&'static str, String), String> {
if a.bytes > MAX_MODULE || a.name.contains('/') || a.name.starts_with('.') {
return Err(format!("[plugin-row] `{}` ({} bytes) is not a plugin this box takes", a.name, a.bytes));
}
let key = object_key(a.blake3.trim(), &a.name);
let bytes = fetch.get(&key, a.bytes as usize).map_err(|e| format!("[plugin-fetch] {key}: {e}"))?;
// The list's digest, BEFORE anything else reads the bytes.
if bytes.len() as u64 != a.bytes {
return Err(format!("[plugin-digest] {key} is {} bytes, the signed list says {}", bytes.len(), a.bytes));
}
let blake3 = monetize_product_wasm::digest_hex(&bytes);
if !blake3.eq_ignore_ascii_case(a.blake3.trim()) || !sha256_hex(&bytes).eq_ignore_ascii_case(a.sha256.trim()) {
return Err(format!("[plugin-digest] {key} hashes to blake3 {blake3}, not the signed list's {}", a.blake3.trim()));
}
let p7s = fetch.get(&format!("{key}.p7s"), MAX_SIGNATURE).map_err(|e| format!("[plugin-unsigned] {key}.p7s: {e}"))?;
let (kind, name) = monetize_product_wasm::identify(&bytes).map_err(|e| format!("[plugin-unreadable] {e}"))?;
let mine: Vec<&Slot> = slots.iter().filter(|s| monetize_product_guest::kinds::loads(s.kind, &kind) && s.name == name).collect();
if mine.is_empty() {
return Err(format!("[plugin-no-block] no block on this box runs the {kind} guest `{name}`"));
}
// Every block's own roots, BEFORE a byte is written anywhere a block names.
let mut refused = Vec::new();
let mut signers = Vec::new();
for s in &mine {
match verify_signed(stage, &a.name, &bytes, &p7s, &s.roots) {
Ok(by) => signers.push(by),
Err(e) => refused.push(format!("{} [plugin-unsigned] {e}", s.block)),
}
}
if !refused.is_empty() {
return Err(refused.join("; "));
}
let by = signers.first().cloned().unwrap_or_default();
let mut changed = Vec::new();
let mut same = Vec::new();
for s in &mine {
let sig = guest_host::ModuleTrust::signature_path(&s.module);
let current = std::fs::read(&s.module).ok();
let current_sig = std::fs::read(&sig).ok();
if current.as_deref() == Some(&bytes[..]) && current_sig.as_deref() == Some(&p7s[..]) {
same.push(s.block.clone());
continue;
}
if apply {
put(&sig, &p7s).and_then(|_| put(&s.module, &bytes)).map_err(|e| format!("{} [plugin-write] {e}", s.block))?;
}
changed.push(format!("{} ({})", s.block, s.module.display()));
}
if changed.is_empty() {
return Ok((UNCHANGED, format!("{} run blake3 {:.12} already", same.join(", "), blake3)));
}
let what = if apply { REPLACED } else { WOULD_REPLACE };
Ok((what, format!("{} ← blake3 {:.12}, signed by {by}", changed.join(", "), blake3)))
}
// ── the release bucket, over HTTPS ──────────────────────────────────────────
/// Pure: **the release bucket's base** (`https://<endpoint>/<bucket>`) from any release
/// object's url this box is configured with (`…/release/<kind>/<file>`); the bench's loopback
/// bucket too.
pub fn bucket_base(object_url: &str) -> Option<String> {
let u = object_url.split('?').next()?;
let i = u.rfind("/release/")?;
(u.starts_with("https://") || u.starts_with("http://127.0.0.1:")).then(|| u[..i].to_string())
}
/// Pure: the bucket the box's configuration names (the release media's objects).
pub fn bucket_of(cfg: &Config) -> Result<String, String> {
let orders = cfg.estate_orders.as_ref().ok_or("[plugins-no-bucket] this box names no release bucket ([estate_orders] is absent)")?;
orders
.media_objects
.values()
.map(|o| o.url.clone())
.chain(cfg.estates.values().flat_map(|e| e.orders.media_objects.values().map(|o| o.url.clone())))
.find_map(|u| bucket_base(&u))
.ok_or_else(|| "[plugins-no-bucket] no release object names the bucket (`[estate_orders.media_objects]`)".to_string())
}
/// **The release bucket this box reads its media from**, each object presigned with the box's
/// own READ pair (`secret.media_objsto`, fed by the kit) when it holds one.
pub struct Bucket {
pub base: String,
pub key: Option<crate::estate_order::ObjstoKey>,
}
impl Bucket {
pub fn of(cfg: &Config) -> Result<Bucket, String> {
let base = bucket_of(cfg)?;
let file = cfg.secrets_dir().join(crate::estate_order::MEDIA_OBJSTO_FED);
let key = match std::fs::read(&file) {
Ok(b) => Some(serde_json::from_slice(&b).map_err(|e| format!("[plugins-bucket-key] the READ pair: {e}"))?),
Err(_) => None,
};
Ok(Bucket { base, key })
}
fn url(&self, object: &str) -> Result<String, String> {
let plain = format!("{}/{object}", self.base);
let url = match &self.key {
#[cfg(feature = "custody-ring")]
Some(k) => {
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
crate::estate_order::presign_get(&plain, k, now, 600)?
}
#[cfg(not(feature = "custody-ring"))]
Some(_) => return Err("[plugins-bucket-key] this build cannot presign (custody-ring is off)".into()),
None => plain,
};
// A guest's own loopback holds no bucket: the bench's mock answers at 10.0.2.100.
Ok(match url.strip_prefix("http://127.0.0.1:") {
Some(rest) => format!("http://10.0.2.100:{rest}"),
None => url,
})
}
}
impl Fetch for Bucket {
fn get(&self, object: &str, cap: usize) -> Result<Vec<u8>, String> {
let url = self.url(object)?;
let _ = rustls::crypto::ring::default_provider().install_default();
let client = reqwest::blocking::Client::builder()
.tls_certs_only(webpki_root_certs::TLS_SERVER_ROOT_CERTS.iter().filter_map(|c| reqwest::Certificate::from_der(c).ok()))
.connect_timeout(std::time::Duration::from_secs(30))
.timeout(std::time::Duration::from_secs(600))
.build()
.map_err(|e| format!("no HTTPS client: {e}"))?;
let resp = client.get(url).send().map_err(|e| e.without_url().to_string())?;
if !resp.status().is_success() {
return Err(format!("the bucket answered {}", resp.status()));
}
if resp.content_length().is_some_and(|n| n as usize > cap) {
return Err(format!("larger than {cap} bytes"));
}
let bytes = resp.bytes().map_err(|e| e.without_url().to_string())?;
if bytes.len() > cap {
return Err(format!("larger than {cap} bytes"));
}
Ok(bytes.to_vec())
}
}
// ── the queued job ──────────────────────────────────────────────────────────
/// The jera job kind of a plugin update.
pub const JOB_KIND: &str = "monetize_plugins";
/// The chain lane's supersede group.
pub const JOB_GROUP: &str = "plugins";
/// Where one job stands.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct JobView {
pub job: String,
/// `queued` | `running` | `done` | `failed`
pub state: String,
pub apply: bool,
pub rows: Vec<Row>,
pub error: String,
}
type Jobs = Arc<Mutex<(BTreeMap<String, JobView>, Option<String>)>>;
/// **The plugin updates of this server**: one jera [`jera::jobs::chain::ChainQueue`] lane
/// (started on the first submit, on the caller's runtime), and what each job came to.
#[derive(Default)]
pub struct PluginJobs {
queue: OnceLock<Arc<jera::jobs::chain::ChainQueue>>,
jobs: Jobs,
/// The tests' bucket; the box's own is read from its configuration per job.
pub fetch: OnceLock<Arc<dyn Fetch>>,
}
/// How many finished jobs are kept to be read back.
pub const KEEP_JOBS: usize = 32;
fn view_of(rec: &jera::jobs::JobRecord) -> JobView {
let detail: serde_json::Value = serde_json::from_str(&rec.detail_json).unwrap_or_default();
JobView {
job: rec.job_id.clone(),
state: rec.status.clone(),
apply: detail.get("apply").and_then(|v| v.as_bool()).unwrap_or(false),
rows: detail.get("rows").and_then(|r| serde_json::from_value(r.clone()).ok()).unwrap_or_default(),
error: detail.get("error").and_then(|v| v.as_str()).unwrap_or_default().to_string(),
}
}
impl PluginJobs {
/// **Queue a look (`apply` false) or an update.** Answers the job at once; the same ask
/// while one is queued or running is that job.
pub fn submit(&self, state: &Arc<crate::state::AppState>, apply: bool) -> Result<JobView, String> {
let queue = self.queue_for(state)?;
let spec = jera::jobs::chain::ChainSpec::new(JOB_KIND, format!("plugins-update-apply={apply}"), JOB_GROUP, 0, "monetize", serde_json::json!({ "apply": apply }))
.with_target(if apply { "update plugins" } else { "look at plugins" })
.with_priority(jera::jobs::chain::Priority::Normal);
let id = queue.submit(spec).job_id().map(str::to_string).ok_or("[plugins-queue] the lane refused the job")?;
Ok(self.status(&id).unwrap_or(JobView { job: id, state: "queued".into(), apply, ..Default::default() }))
}
/// One job; the newest for an empty `job`.
pub fn status(&self, job: &str) -> Option<JobView> {
let j = self.jobs.lock().unwrap_or_else(|e| e.into_inner());
let id = if job.trim().is_empty() { j.1.clone()? } else { job.trim().to_string() };
j.0.get(&id).cloned()
}
fn queue_for(&self, state: &Arc<crate::state::AppState>) -> Result<Arc<jera::jobs::chain::ChainQueue>, String> {
if let Some(q) = self.queue.get() {
return Ok(q.clone());
}
let rt = tokio::runtime::Handle::try_current().map_err(|_| "[plugins-queue] no runtime to run the lane on".to_string())?;
let jobs = self.jobs.clone();
let sink = jera::jobs::JobSink::remote(move |rec: &jera::jobs::JobRecord| {
// Every plugin update is a row in the server's ledger too (RA4-6; they are rare).
crate::jobs::ledger_sink().record(rec);
let mut j = jobs.lock().unwrap_or_else(|e| e.into_inner());
let mut v = view_of(rec);
if let Some(old) = j.0.get(&rec.job_id) {
if v.rows.is_empty() && v.error.is_empty() {
v.apply = old.apply || v.apply;
}
}
if !j.0.contains_key(&rec.job_id) {
j.1 = Some(rec.job_id.clone());
}
j.0.insert(rec.job_id.clone(), v);
while j.0.len() > KEEP_JOBS {
let Some(oldest) = j.0.keys().next().cloned() else { break };
if Some(&oldest) == j.1.as_ref() {
break;
}
j.0.remove(&oldest);
}
});
let queue = jera::jobs::chain::ChainQueue::new(sink);
let weak: Weak<crate::state::AppState> = Arc::downgrade(state);
let q = self.queue.get_or_init(|| queue.clone()).clone();
if Arc::ptr_eq(&q, &queue) {
// A32-14, stays a task: the plugin lane's jera worker.
rt.spawn(queue.clone().run(move |spec| run_job(&weak, &spec.payload)));
}
Ok(q)
}
}
/// The job itself: on the lane's blocking thread.
fn run_job(state: &Weak<crate::state::AppState>, payload: &serde_json::Value) -> anyhow::Result<serde_json::Value> {
let apply = payload.get("apply").and_then(|v| v.as_bool()).unwrap_or(false);
let Some(state) = state.upgrade() else { anyhow::bail!("[plugins-gone] the server stopped") };
let cfg = state.config.read().unwrap_or_else(|e| e.into_inner()).clone();
let stage = cfg.data_dir.join("plugin-update");
let fetch: Arc<dyn Fetch> = match state.plugin_jobs.fetch.get() {
Some(f) => f.clone(),
None => match Bucket::of(&cfg) {
Ok(b) => Arc::new(b),
Err(e) => return Ok(serde_json::json!({ "apply": apply, "error": e })),
},
};
let out = match update(&cfg, fetch.as_ref(), apply, &stage) {
Ok(rows) => {
for r in &rows {
let line = format!("plugin {} [{}] {}: {} {}", if apply { "update" } else { "look" }, r.role, r.name, r.outcome, r.detail);
if r.outcome == REFUSED {
eprintln!("monetize-server: ALERT plugin.refused {line}");
} else {
eprintln!("monetize-server: {line}");
}
crate::guests::journal_push(&line);
}
serde_json::json!({ "apply": apply, "rows": rows })
}
Err(e) => {
eprintln!("monetize-server: ALERT plugin.refused plugin update REFUSED {e}");
serde_json::json!({ "apply": apply, "error": e })
}
};
let _ = std::fs::remove_dir_all(&stage);
Ok(out)
}