299 lines
14 KiB
Rust
299 lines
14 KiB
Rust
//! The plugin directory: plugin bundles (`figcraft_plugin::bundle`) published to the server with
|
||
//! its key, listed, searched and fetched by anyone who can reach it, and unpublished with the key
|
||
//! (there are no accounts: whoever has the key is the publisher, and the operator moderates).
|
||
//!
|
||
//! ```text
|
||
//! <data>/plugins/index.json every version kept: manifest fields, size, SHA-256, when
|
||
//! <data>/plugins/<sha256>.bin a bundle, named by the SHA-256 of its bytes
|
||
//! ```
|
||
//!
|
||
//! A bundle is checked in full before it is kept (it decodes within the size cap, its manifest
|
||
//! parses and asks for a plugin API this build runs, its version is newer than the one kept, its
|
||
//! module compiles and imports only what the sandbox provides, its asset names are plain) and is
|
||
//! never run here. Bundles are named by their hash, so no name a publisher picks reaches the file
|
||
//! system, and a fetch reads the file back and checks its hash before sending it. Caps: bundle
|
||
//! size, versions kept and bytes kept; fetches share a budget of bytes per second, so a crowd of
|
||
//! downloads can't queue more than a bounded amount of data.
|
||
|
||
use std::cmp::Ordering;
|
||
use std::collections::BTreeMap;
|
||
use std::fs;
|
||
use std::path::{Path, PathBuf};
|
||
use std::time::Instant;
|
||
|
||
use figcraft_collab::message::{PluginEntry, PluginReply, PluginReq};
|
||
use figcraft_collab_net::net::Bucket;
|
||
use figcraft_plugin::bundle::{Bundle, compare_versions, parse_version, sha256_hex};
|
||
use figcraft_plugin::manifest::Access;
|
||
|
||
use crate::disk;
|
||
use crate::store::now;
|
||
|
||
pub const PLUGINS: &str = "plugins";
|
||
pub const INDEX: &str = "index.json";
|
||
/// Largest `index.json` read, longest search, and most bytes of entries one listing carries.
|
||
const MAX_INDEX_FILE: u64 = 64 << 20;
|
||
const MAX_QUERY: usize = 200;
|
||
const MAX_LISTING: usize = 32 << 20;
|
||
/// What fetches may send, in MiB per second on average (each costs at least one).
|
||
const SERVE_MIB_PER_SEC: f64 = 32.0;
|
||
/// …and at once: eight times what one block of addresses may download at once, so no block can
|
||
/// take it all (see `abuse`).
|
||
const SERVE_BURST_MIB: f64 = 512.0;
|
||
const MIB: f64 = (1 << 20) as f64;
|
||
|
||
/// What the directory enforces.
|
||
#[derive(Clone, Copy, Debug, PartialEq)]
|
||
pub struct DirectoryConfig {
|
||
/// Most versions kept (all plugins together); 0 turns the directory off.
|
||
pub max_plugins: usize,
|
||
/// Largest bundle.
|
||
pub max_plugin_bytes: u64,
|
||
/// Most bytes of bundles kept.
|
||
pub max_total_bytes: u64,
|
||
}
|
||
|
||
/// The plugins a server keeps.
|
||
pub struct Directory {
|
||
dir: PathBuf,
|
||
cfg: DirectoryConfig,
|
||
entries: Vec<PluginEntry>,
|
||
served: Bucket,
|
||
/// `index.json` couldn't be read: nothing is written over it until the operator fixes it.
|
||
broken: bool,
|
||
}
|
||
|
||
/// A plugin id as manifests have them (letters, digits, `.`, `-`, `_`; 1–128).
|
||
fn valid_id(s: &str) -> bool {
|
||
(1..=128).contains(&s.len()) && s.bytes().all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-' | b'_'))
|
||
}
|
||
|
||
fn valid_sha(s: &str) -> bool {
|
||
s.len() == 64 && s.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
|
||
}
|
||
|
||
/// Newest first.
|
||
fn by_version(a: &PluginEntry, b: &PluginEntry) -> Ordering {
|
||
compare_versions(&b.version, &a.version).unwrap_or(Ordering::Equal)
|
||
}
|
||
|
||
impl Directory {
|
||
/// The directory under `root` (made when the first plugin is published). Entries whose bundle
|
||
/// is missing or the wrong size are left out and reported, never deleted.
|
||
pub fn open(root: &Path, cfg: DirectoryConfig) -> (Directory, Vec<String>) {
|
||
let dir = root.join(PLUGINS);
|
||
let burst = (2 * cfg.max_plugin_bytes) as f64 / MIB;
|
||
let mut d = Directory { dir, cfg, entries: Vec::new(), served: Bucket::new(SERVE_MIB_PER_SEC, burst.max(SERVE_BURST_MIB)), broken: false };
|
||
let mut problems = Vec::new();
|
||
let index = d.dir.join(INDEX);
|
||
let _ = fs::remove_file(disk::temp_of(&index));
|
||
if !index.exists() {
|
||
return (d, problems);
|
||
}
|
||
let all: Vec<PluginEntry> =
|
||
match disk::read_capped(&index, MAX_INDEX_FILE).and_then(|b| serde_json::from_slice(&b).map_err(|e| e.to_string())) {
|
||
Ok(all) => all,
|
||
Err(e) => {
|
||
problems.push(format!("plugin directory: {INDEX} can't be read ({e}); no plugins are listed or published until it is fixed"));
|
||
d.broken = true;
|
||
return (d, problems);
|
||
}
|
||
};
|
||
for e in all {
|
||
let size = fs::metadata(d.file(&e.sha256)).map(|m| m.len()).ok();
|
||
if !valid_id(&e.id) || parse_version(&e.version).is_none() || !valid_sha(&e.sha256) || size != Some(e.bytes) {
|
||
problems.push(format!(
|
||
"plugin directory: skipped {} {}: its bundle is missing or damaged",
|
||
e.id.escape_debug(),
|
||
e.version.escape_debug()
|
||
));
|
||
continue;
|
||
}
|
||
d.entries.push(e);
|
||
}
|
||
(d, problems)
|
||
}
|
||
|
||
fn file(&self, sha: &str) -> PathBuf {
|
||
self.dir.join(format!("{sha}.bin"))
|
||
}
|
||
|
||
/// Every version kept.
|
||
pub fn entries(&self) -> &[PluginEntry] {
|
||
&self.entries
|
||
}
|
||
|
||
/// Bytes of bundles kept (a bundle published under two versions counts once).
|
||
fn total_bytes(&self) -> u64 {
|
||
let files: BTreeMap<&str, u64> = self.entries.iter().map(|e| (e.sha256.as_str(), e.bytes)).collect();
|
||
files.values().sum()
|
||
}
|
||
|
||
/// Answer one request; `publisher`: the connection has the server's key.
|
||
pub fn handle(&mut self, publisher: bool, req: PluginReq) -> PluginReply {
|
||
let failed = |reason: String| PluginReply::Failed { reason };
|
||
if self.cfg.max_plugins == 0 {
|
||
return failed("this server's plugin directory is turned off".into());
|
||
}
|
||
let needs_key = || failed("publishing and unpublishing plugins needs the server's key (ws://host:port/?key=…)".into());
|
||
match req {
|
||
PluginReq::List { query } => PluginReply::List(self.list(&query)),
|
||
PluginReq::Fetch { id, version } => match self.fetch(&id, version.as_deref()) {
|
||
Ok((entry, bundle)) => PluginReply::Bundle { entry, bundle },
|
||
Err(e) => failed(e),
|
||
},
|
||
PluginReq::Publish { .. } | PluginReq::Unpublish { .. } if !publisher => needs_key(),
|
||
PluginReq::Publish { bundle } => self.publish(&bundle).map_or_else(failed, PluginReply::Published),
|
||
PluginReq::Unpublish { id, version } => self.unpublish(&id, version.as_deref()).map_or_else(failed, PluginReply::Removed),
|
||
}
|
||
}
|
||
|
||
/// The latest version of each plugin matching `query` (any case, in its id, name, description
|
||
/// or author), by name, with the versions kept.
|
||
pub fn list(&self, query: &str) -> Vec<PluginEntry> {
|
||
let q: String = query.trim().chars().take(MAX_QUERY).collect::<String>().to_lowercase();
|
||
let mut by_id: BTreeMap<&str, Vec<&PluginEntry>> = BTreeMap::new();
|
||
for e in &self.entries {
|
||
by_id.entry(e.id.as_str()).or_default().push(e);
|
||
}
|
||
let mut out: Vec<PluginEntry> = by_id
|
||
.into_values()
|
||
.filter_map(|mut versions| {
|
||
versions.sort_by(|a, b| by_version(a, b));
|
||
let mut latest = (*versions.first()?).clone();
|
||
latest.versions = versions.iter().map(|v| v.version.clone()).collect();
|
||
Some(latest)
|
||
})
|
||
.filter(|e| q.is_empty() || [&e.id, &e.name, &e.description, &e.author].iter().any(|f| f.to_lowercase().contains(&q)))
|
||
.collect();
|
||
out.sort_by(|a, b| a.name.to_lowercase().cmp(&b.name.to_lowercase()).then_with(|| a.id.cmp(&b.id)));
|
||
// A listing stays well within a message.
|
||
let mut size = 0;
|
||
out.retain(|e| {
|
||
size += 256 + e.name.len() + e.description.len() + e.author.len() + e.commands.iter().map(|c| c.len() + 8).sum::<usize>();
|
||
size <= MAX_LISTING
|
||
});
|
||
out
|
||
}
|
||
|
||
/// The size of version `version` (or the latest) of plugin `id`'s bundle, without reading it.
|
||
pub fn size_of(&self, id: &str, version: Option<&str>) -> Option<u64> {
|
||
self.entries
|
||
.iter()
|
||
.filter(|e| e.id == id && version.is_none_or(|v| compare_versions(&e.version, v) == Some(Ordering::Equal)))
|
||
.min_by(|a, b| by_version(a, b))
|
||
.map(|e| e.bytes)
|
||
}
|
||
|
||
/// Version `version` (or the latest) of plugin `id`, and its bundle, read back and checked
|
||
/// against its hash.
|
||
pub fn fetch(&mut self, id: &str, version: Option<&str>) -> Result<(PluginEntry, Vec<u8>), String> {
|
||
let entry = self
|
||
.entries
|
||
.iter()
|
||
.filter(|e| e.id == id && version.is_none_or(|v| compare_versions(&e.version, v) == Some(Ordering::Equal)))
|
||
.min_by(|a, b| by_version(a, b))
|
||
.cloned()
|
||
.ok_or_else(|| match version {
|
||
Some(v) => format!("the directory has no version {} of that plugin", v.chars().take(32).collect::<String>()),
|
||
None => "the directory has no such plugin (it may have been unpublished)".to_string(),
|
||
})?;
|
||
if !self.served.take_n(Instant::now(), 1.0 + entry.bytes as f64 / MIB) {
|
||
return Err("the server is busy sending plugins: try again in a moment".into());
|
||
}
|
||
let bytes = disk::read_capped(&self.file(&entry.sha256), self.cfg.max_plugin_bytes.max(entry.bytes))?;
|
||
if sha256_hex(&bytes) != entry.sha256 {
|
||
log::error!("plugin directory: the bundle of {} {} is damaged on disk; not sending it", entry.id, entry.version);
|
||
return Err("that plugin's bundle is damaged on the server; tell the server's owner".into());
|
||
}
|
||
Ok((entry, bytes))
|
||
}
|
||
|
||
/// Keep a bundle as a new version of its plugin, after checking all of it.
|
||
// shortcut: the bundle is checked (its module compiled) on the hub thread, like team libraries;
|
||
// move it to a worker if large plugins make other sessions wait noticeably.
|
||
pub fn publish(&mut self, bytes: &[u8]) -> Result<PluginEntry, String> {
|
||
let max = self.cfg.max_plugin_bytes;
|
||
if bytes.len() as u64 > max {
|
||
return Err(format!("the plugin is {} KiB; this server keeps plugins up to {} KiB", bytes.len() >> 10, max >> 10));
|
||
}
|
||
let bundle = Bundle::decode(bytes, usize::try_from(max).unwrap_or(usize::MAX))?;
|
||
let m = bundle.check()?;
|
||
if let Some(latest) = self.entries.iter().filter(|e| e.id == m.id).min_by(|a, b| by_version(a, b))
|
||
&& compare_versions(&m.version, &latest.version) != Some(Ordering::Greater)
|
||
{
|
||
return Err(format!("version {} of {} is on the server already: publish a newer version", latest.version, m.id));
|
||
}
|
||
if self.entries.len() >= self.cfg.max_plugins {
|
||
return Err(format!("the directory keeps at most {} plugin versions: unpublish some first", self.cfg.max_plugins));
|
||
}
|
||
let sha = sha256_hex(bytes);
|
||
let known = self.entries.iter().any(|e| e.sha256 == sha);
|
||
if !known && self.total_bytes() + bytes.len() as u64 > self.cfg.max_total_bytes {
|
||
return Err(format!("the directory is full ({} MiB): unpublish some plugins first", self.cfg.max_total_bytes >> 20));
|
||
}
|
||
let entry = PluginEntry {
|
||
id: m.id.clone(),
|
||
name: m.name.clone(),
|
||
version: m.version.clone(),
|
||
description: m.description.clone(),
|
||
author: m.author.clone(),
|
||
api: m.api,
|
||
access: match m.permissions.access {
|
||
Access::Read => "read".into(),
|
||
Access::Write => "write".into(),
|
||
},
|
||
commands: m.permissions.commands.clone(),
|
||
widget: m.widget,
|
||
bytes: bytes.len() as u64,
|
||
sha256: sha.clone(),
|
||
published: now(),
|
||
versions: Vec::new(),
|
||
};
|
||
fs::create_dir_all(&self.dir).map_err(|e| format!("couldn't keep the plugin: {e}"))?;
|
||
let file = self.file(&sha);
|
||
if !known {
|
||
disk::write_atomic(&file, bytes).map_err(|e| format!("couldn't keep the plugin: {e}"))?;
|
||
}
|
||
self.entries.push(entry.clone());
|
||
if let Err(e) = self.save() {
|
||
self.entries.pop();
|
||
return Err(format!("couldn't keep the plugin: {e}"));
|
||
}
|
||
log::info!("plugin directory: published {} {} ({} bytes)", entry.id, entry.version, entry.bytes);
|
||
Ok(entry)
|
||
}
|
||
|
||
/// Remove version `version` (or every version) of plugin `id`; the versions removed.
|
||
pub fn unpublish(&mut self, id: &str, version: Option<&str>) -> Result<Vec<String>, String> {
|
||
let gone = |e: &PluginEntry| e.id == id && version.is_none_or(|v| compare_versions(&e.version, v) == Some(Ordering::Equal));
|
||
let (removed, kept): (Vec<PluginEntry>, Vec<PluginEntry>) = self.entries.iter().cloned().partition(gone);
|
||
if removed.is_empty() {
|
||
return Err("the directory has no such plugin or version".into());
|
||
}
|
||
let before = std::mem::replace(&mut self.entries, kept);
|
||
if let Err(e) = self.save() {
|
||
self.entries = before;
|
||
return Err(format!("couldn't unpublish: {e}"));
|
||
}
|
||
for r in &removed {
|
||
if !self.entries.iter().any(|e| e.sha256 == r.sha256)
|
||
&& let Err(e) = fs::remove_file(self.file(&r.sha256))
|
||
{
|
||
log::warn!("plugin directory: couldn't delete the bundle of {} {}: {e}", r.id, r.version);
|
||
}
|
||
}
|
||
let versions: Vec<String> = removed.into_iter().map(|e| e.version).collect();
|
||
log::info!("plugin directory: unpublished {id} {}", versions.join(", "));
|
||
Ok(versions)
|
||
}
|
||
|
||
fn save(&self) -> Result<(), String> {
|
||
if self.broken {
|
||
return Err(format!("the directory's {INDEX} is damaged; tell the server's owner"));
|
||
}
|
||
let text = serde_json::to_vec_pretty(&self.entries).map_err(|e| e.to_string())?;
|
||
disk::write_atomic(&self.dir.join(INDEX), &text).map_err(|e| e.to_string())
|
||
}
|
||
}
|