From 45c3b96907f62399a6699db91ef6212585c61a24 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Mon, 20 Jul 2026 20:12:18 +0200 Subject: [PATCH] =?UTF-8?q?feat(store):=20plugin=20store=20host=20module?= =?UTF-8?q?=20=E2=80=94=20signed=20catalogs,=20tiered=20trust,=20install?= =?UTF-8?q?=20jobs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The index is the verification gate: a catalog entry pins one exact version plus that version's tarball integrity hash, so 'verified on every release' is a property of the data rather than a promise about process. Nothing can express 'track latest' for a catalogued plugin. - store/index.rs signed index parse + ed25519 verify (ring), validate-and-drop per entry, semver minHost/advisory ranges - store/sources.rs built-in unom source (compiled-in URL + two key slots for rotation) + operator sources in plugin-sources.json - store/catalog.rs https fetch with size/timeout/redirect caps, signature before parse, last-good disk cache (stale-but-usable when offline) - store/jobs.rs single-flight install/uninstall: registry-integrity preflight against the pin, spawn the runner CLI with live log capture, post-install version check with rollback, provenance record, runner restart (discovery is startup-only) - store/manifest.rs install provenance; absence means CLI-installed - mgmt/store.rs 12 routes under /api/v1/store, denied to the plugin token (a plugin that can install plugins is an escalation primitive) Also generalizes runner discovery and listInstalled from @punktfunk/plugin-* to ANY scope's plugin-*: catalog entries must be scoped so the scope can map to that entry's registry, so a third-party plugin necessarily arrives under its own scope and would otherwise install but never run. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 2 + crates/punktfunk-host/Cargo.toml | 7 + crates/punktfunk-host/src/events.rs | 7 + crates/punktfunk-host/src/main.rs | 3 + crates/punktfunk-host/src/mgmt.rs | 14 +- crates/punktfunk-host/src/mgmt/auth.rs | 7 + crates/punktfunk-host/src/mgmt/plugins.rs | 19 + crates/punktfunk-host/src/mgmt/store.rs | 755 ++++++++++++++++++++ crates/punktfunk-host/src/plugins.rs | 194 ++++- crates/punktfunk-host/src/store.rs | 394 ++++++++++ crates/punktfunk-host/src/store/catalog.rs | 255 +++++++ crates/punktfunk-host/src/store/index.rs | 570 +++++++++++++++ crates/punktfunk-host/src/store/jobs.rs | 687 ++++++++++++++++++ crates/punktfunk-host/src/store/manifest.rs | 246 +++++++ crates/punktfunk-host/src/store/sources.rs | 286 ++++++++ sdk/src/plugins.ts | 84 ++- sdk/src/runner-cli.ts | 49 +- sdk/src/runner.ts | 15 +- sdk/test/plugins.test.ts | 49 ++ 19 files changed, 3578 insertions(+), 65 deletions(-) create mode 100644 crates/punktfunk-host/src/mgmt/store.rs create mode 100644 crates/punktfunk-host/src/store.rs create mode 100644 crates/punktfunk-host/src/store/catalog.rs create mode 100644 crates/punktfunk-host/src/store/index.rs create mode 100644 crates/punktfunk-host/src/store/jobs.rs create mode 100644 crates/punktfunk-host/src/store/manifest.rs create mode 100644 crates/punktfunk-host/src/store/sources.rs diff --git a/Cargo.lock b/Cargo.lock index dd2d802c..ddf40320 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3363,11 +3363,13 @@ dependencies = [ "rand 0.8.6", "rcgen", "reis", + "ring", "roxmltree", "rsa", "rusqlite", "rustls", "rusty_enet", + "semver", "serde", "serde_json", "sha2", diff --git a/crates/punktfunk-host/Cargo.toml b/crates/punktfunk-host/Cargo.toml index 04204fdc..b1ba5d57 100644 --- a/crates/punktfunk-host/Cargo.toml +++ b/crates/punktfunk-host/Cargo.toml @@ -110,6 +110,13 @@ serde_json = "1" # utoipa into axum 0.8 extractors; utoipa-axum collects `#[utoipa::path]` routes into the # spec; utoipa-scalar serves the interactive docs. Codegen-friendly: the spec is emitted # verbatim by the `openapi` subcommand. Control plane only — never the per-frame path. +# Plugin-store index signatures: ed25519 verification of a catalog document before any field of +# it is read (store/index.rs). Already in the tree via rustls — the workspace is ring-only, so this +# is the one signature primitive available without pulling aws-lc-sys (which fails on Windows CI). +ring = "0.17" +# Semver comparisons for the plugin store: `minHost` gating and the revocation list's version +# ranges (`<0.3.2`). Already in the lockfile transitively. +semver = "1" utoipa = { version = "5", features = ["axum_extras"] } utoipa-axum = "0.2" utoipa-scalar = { version = "0.3", features = ["axum"] } diff --git a/crates/punktfunk-host/src/events.rs b/crates/punktfunk-host/src/events.rs index fa309451..32f4ac15 100644 --- a/crates/punktfunk-host/src/events.rs +++ b/crates/punktfunk-host/src/events.rs @@ -170,6 +170,12 @@ pub enum EventKind { /// lease-expired). A consumer re-reads `GET /api/v1/plugins` for the new set. id: String, }, + #[serde(rename = "store.changed")] + /// The set of installed plugins, or what the store knows about them, changed — an install or + /// uninstall finished, or a catalog refresh brought in new rows. A consumer re-reads + /// `GET /api/v1/store/catalog` / `…/installed`. Deliberately payload-free: the store's answer + /// is a join over several sources of truth, so "go look again" is the only honest signal. + StoreChanged, #[serde(rename = "host.started")] HostStarted { version: String, @@ -197,6 +203,7 @@ impl EventKind { EventKind::DisplayReleased { .. } => "display.released", EventKind::LibraryChanged { .. } => "library.changed", EventKind::PluginsChanged { .. } => "plugins.changed", + EventKind::StoreChanged => "store.changed", EventKind::HostStarted { .. } => "host.started", EventKind::HostStopping => "host.stopping", } diff --git a/crates/punktfunk-host/src/main.rs b/crates/punktfunk-host/src/main.rs index 11607fd8..f2159707 100644 --- a/crates/punktfunk-host/src/main.rs +++ b/crates/punktfunk-host/src/main.rs @@ -74,6 +74,9 @@ mod session_plan; mod session_status; mod spike; mod stats_recorder; +// The plugin store: signed catalogs, tiered trust, and install/uninstall jobs that run through the +// same runner CLI the `plugins` subcommand uses (design/plugin-store.md). +mod store; mod stream_marker; // `monitor_devnode::startup_recover()` (below) re-enables PnP monitor devnodes disabled by a prior // run; it lives in the `pf-win-display` leaf crate (plan §W6). diff --git a/crates/punktfunk-host/src/mgmt.rs b/crates/punktfunk-host/src/mgmt.rs index 8433ae8e..4dcf9336 100644 --- a/crates/punktfunk-host/src/mgmt.rs +++ b/crates/punktfunk-host/src/mgmt.rs @@ -42,6 +42,7 @@ mod plugins; mod session; mod shared; mod stats; +mod store; #[cfg(test)] mod tests; @@ -240,7 +241,17 @@ fn api_router_parts() -> (Router>, utoipa::openapi::OpenApi) { .routes(routes!(hooks::get_hooks, hooks::set_hooks)) .routes(routes!(plugins::list_plugins)) .routes(routes!(plugins::register_plugin, plugins::delete_plugin)) - .routes(routes!(plugins::get_ui_credential)), + .routes(routes!(plugins::get_ui_credential)) + .routes(routes!(store::get_catalog)) + .routes(routes!(store::refresh_catalog)) + .routes(routes!(store::list_installed)) + .routes(routes!(store::install_plugin)) + .routes(routes!(store::uninstall_plugin)) + .routes(routes!(store::list_jobs)) + .routes(routes!(store::get_job)) + .routes(routes!(store::list_sources)) + .routes(routes!(store::put_source, store::delete_source)) + .routes(routes!(store::get_runtime, store::set_runtime)), ) .split_for_parts() } @@ -279,6 +290,7 @@ pub fn openapi_json() -> String { (name = "events", description = "Host lifecycle events: an SSE stream (client/session/stream lifecycle, pairing, displays, library, host) with Last-Event-ID resume and server-side kind filters"), (name = "hooks", description = "Operator hooks: commands and webhooks fired on lifecycle events (fire-and-forget — hooks observe, never veto)"), (name = "plugins", description = "Plugin directory: running `punktfunk-plugin-*` processes register a lease and, optionally, a loopback UI the web console proxies and adds to its nav"), + (name = "store", description = "Plugin store: browse signed catalogs (verified first-party entries, attributed third-party sources), install/uninstall as tracked jobs, and switch the plugin runner on"), ) )] struct ApiDoc; diff --git a/crates/punktfunk-host/src/mgmt/auth.rs b/crates/punktfunk-host/src/mgmt/auth.rs index 2ac555d8..324eda16 100644 --- a/crates/punktfunk-host/src/mgmt/auth.rs +++ b/crates/punktfunk-host/src/mgmt/auth.rs @@ -131,8 +131,15 @@ pub(crate) async fn require_auth( /// or eject the operator's. /// - **UI proxy credentials** — a plugin has no business reading another plugin's per-boot UI /// secret; only the console proxy (admin token) needs it. +/// - **the plugin store** — installing a plugin is running new code with operator privileges, and a +/// plugin that can do that is a persistence/escalation primitive: it could install a helper that +/// isn't constrained the way it is, or switch the runner's own service state. Denied wholesale +/// (reads included — the catalog is not sensitive, but there is no reason a plugin needs it, and +/// a whole-prefix deny can't be defeated by a route added later). pub(crate) fn plugin_may_access(method: &Method, path: &str) -> bool { let denied = path == "/api/v1/hooks" + || path == "/api/v1/store" + || path.starts_with("/api/v1/store/") || path == "/api/v1/pair" || path.starts_with("/api/v1/pair/") || path == "/api/v1/native/pair" diff --git a/crates/punktfunk-host/src/mgmt/plugins.rs b/crates/punktfunk-host/src/mgmt/plugins.rs index 46f76756..992e857e 100644 --- a/crates/punktfunk-host/src/mgmt/plugins.rs +++ b/crates/punktfunk-host/src/mgmt/plugins.rs @@ -202,6 +202,15 @@ impl PluginRegistry { }) } + /// The ids of live registrations, without pruning or announcing anything. + fn live_ids(&self) -> Vec { + let map = self.inner.read().unwrap_or_else(|e| e.into_inner()); + map.iter() + .filter(|(_, s)| s.is_live()) + .map(|(id, _)| id.clone()) + .collect() + } + /// Remove `id`. Returns `true` if a **live** entry existed (a clean deregister); removing an /// already-expired or unknown id returns `false` (nothing to announce). fn remove(&self, id: &str) -> bool { @@ -222,6 +231,16 @@ pub(crate) fn registry() -> &'static PluginRegistry { REG.get_or_init(PluginRegistry::new) } +/// Which plugin ids are registered right now — the plugin store's "running" column +/// ([`super::store::list_installed`]). +/// +/// Deliberately **not** [`list_plugins`]: that one prunes expired leases and emits +/// `plugins.changed` for each, which is right for the nav's poll but would turn the store's own +/// polling into an event fire-hose. This is a pure read. +pub(crate) fn live_plugin_ids() -> Vec { + registry().live_ids() +} + // ---------------------------------------------------------------- validation /// A plugin id: `definePlugin`'s kebab-case name (`^[a-z][a-z0-9-]*$`, ≤64) — the same regex the SDK diff --git a/crates/punktfunk-host/src/mgmt/store.rs b/crates/punktfunk-host/src/mgmt/store.rs new file mode 100644 index 00000000..21902a9f --- /dev/null +++ b/crates/punktfunk-host/src/mgmt/store.rs @@ -0,0 +1,755 @@ +//! Plugin **store** API: browse catalogs, install/uninstall, manage sources and the runner +//! (design `plugin-store.md` §4.2). The domain logic lives in [`crate::store`]; this is its HTTP +//! surface. +//! +//! Auth: **admin bearer + loopback**, like every mutation — and explicitly *denied to the plugin +//! token* ([`super::auth::plugin_may_access`]). A plugin that can install plugins is a persistence +//! and escalation primitive: it could install a helper that isn't constrained the way it is. That +//! deny is a carve-out in the exclusion-based allowlist, so it is spelled out there and pinned by a +//! test. +//! +//! (A console *session* can already reach arbitrary command execution via `PUT /hooks`, so the +//! store adds convenience for a session holder rather than a new privilege class. It is still worth +//! keeping the plugin lane away from it — the lanes exist precisely so a plugin defect isn't an +//! operator compromise.) +//! +//! Everything here does blocking work — filesystem scans, network fetches, process spawns — so +//! handlers hop to `spawn_blocking` rather than stalling the runtime. + +use super::shared::*; +use crate::store::{self, index, jobs, manifest, sources}; + +/// Run blocking store work off the async runtime, mapping a joined-thread panic to a 500. +async fn blocking(f: F) -> Result +where + F: FnOnce() -> T + Send + 'static, + T: Send + 'static, +{ + tokio::task::spawn_blocking(f).await.map_err(|e| { + tracing::error!("plugin-store worker panicked: {e}"); + api_error( + StatusCode::INTERNAL_SERVER_ERROR, + "the plugin store worker failed", + ) + }) +} + +// ---------------------------------------------------------------- wire shapes + +/// Facts about this host, so the console can grey out rows it can't install. +#[derive(Serialize, ToSchema)] +pub(crate) struct HostFacts { + pub version: String, + /// `linux` / `windows` / `macos`. + pub platform: String, +} + +/// A configured catalog source and how its last refresh went. +#[derive(Serialize, ToSchema)] +pub(crate) struct SourceView { + pub name: String, + pub url: String, + /// The built-in `unom` source: not editable, not removable, and the only source whose entries + /// may carry the "verified" tier. + pub builtin: bool, + /// Whether we check a signature on this source's index. An unsigned source still works; the + /// console marks it. + pub signed: bool, + /// The catalog we're serving is older than the last refresh attempt (offline, or the last + /// fetch failed) — entries still install, because the pin travelled with the entry. + pub stale: bool, + /// Unix seconds of the data we hold, when we hold any. + #[serde(skip_serializing_if = "Option::is_none")] + pub fetched_at: Option, + /// Why the last refresh failed, if it did. + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + pub entry_count: u32, + #[serde(skip_serializing_if = "Option::is_none")] + pub public_key: Option, +} + +/// One row on the shelf. +#[derive(Serialize, ToSchema)] +pub(crate) struct CatalogEntry { + pub id: String, + pub pkg: String, + pub title: String, + pub description: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub icon: Option, + pub author: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub homepage: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub license: Option, + /// The one installable version this entry pins. + pub version: String, + /// Which source listed it. + pub source: String, + /// `verified` (built-in source) or `external` (an operator-added source). Never `unverified`: + /// unverified installs come from a raw spec and are never listed (D7). + pub tier: String, + /// When unom reviewed this exact tarball (built-in source only). + #[serde(skip_serializing_if = "Option::is_none")] + pub reviewed_at: Option, + pub platforms: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub min_host: Option, + /// Can this host install it? + pub compatible: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub incompatible_reason: Option, + /// The version installed right now, if any. + #[serde(skip_serializing_if = "Option::is_none")] + pub installed_version: Option, + /// Installed, but at a different version than the catalog pins. + pub update_available: bool, + /// A revocation covering the catalogued version — do not offer this without shouting. + #[serde(skip_serializing_if = "Option::is_none")] + pub blocked: Option, +} + +#[derive(Serialize, ToSchema)] +pub(crate) struct CatalogResponse { + pub host: HostFacts, + pub sources: Vec, + pub plugins: Vec, + /// True while a package operation is in flight — the console disables install buttons. + pub busy: bool, +} + +/// An installed plugin package, joined with its provenance and whether it's actually running. +#[derive(Serialize, ToSchema)] +pub(crate) struct InstalledView { + pub pkg: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub version: Option, + /// `verified` / `external` / `unverified` / `cli` — remembered from install time, so an + /// unverified plugin stays visibly unverified long after the dialog is forgotten. + pub tier: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub source: Option, + /// The catalog entry this maps to, when it is on a shelf we know. + #[serde(skip_serializing_if = "Option::is_none")] + pub entry_id: Option, + /// The plugin id it registers under — the key into `GET /plugins`. + #[serde(skip_serializing_if = "Option::is_none")] + pub plugin_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub title: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub installed_at: Option, + /// Is it registered in the live lease registry right now? + pub running: bool, + /// The catalog's version, when it's newer than what's installed. + #[serde(skip_serializing_if = "Option::is_none")] + pub update_available: Option, + /// A revocation covering the *installed* version. Reported, never auto-removed: silently + /// deleting running code is its own hazard, so the operator decides. + #[serde(skip_serializing_if = "Option::is_none")] + pub blocked: Option, +} + +/// `POST /store/install` — either a catalogued entry, or a raw spec the operator owns. +#[derive(Deserialize, ToSchema)] +pub(crate) struct InstallRequest { + /// Catalog source name (with [`Self::id`]). + #[serde(default)] + pub source: Option, + /// Catalog entry id (with [`Self::source`]). + #[serde(default)] + pub id: Option, + /// A raw package spec (`@scope/name`, `@scope/name@1.2.3`, an https tarball or git+https URL). + /// Nothing reviewed it and nothing pins it. + #[serde(default)] + pub spec: Option, + /// Required with [`Self::spec`]: the operator's explicit acknowledgement that this installs + /// unreviewed code with operator privileges. The console collects it behind a typed + /// confirmation; the API refuses without it so no other caller can skip the decision. + #[serde(default)] + pub accept_unverified: bool, +} + +#[derive(Deserialize, ToSchema)] +pub(crate) struct UninstallRequest { + pub pkg: String, +} + +/// 202 body: where to watch the work. +#[derive(Serialize, ToSchema)] +pub(crate) struct JobRef { + pub job: String, +} + +#[derive(Deserialize, ToSchema)] +pub(crate) struct SourceInput { + pub url: String, + /// `ed25519:`. Omitted ⇒ an unsigned source (accepted, flagged everywhere). + #[serde(default)] + pub public_key: Option, +} + +#[derive(Serialize, ToSchema)] +pub(crate) struct RuntimeView { + /// Is the runner payload/unit present at all? + pub installed: bool, + pub enabled: bool, + pub running: bool, + /// systemd unit or scheduled-task name. + pub unit: String, + /// Windows: the account the task runs as. + #[serde(skip_serializing_if = "Option::is_none")] + pub principal: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub detail: Option, +} + +#[derive(Deserialize, ToSchema)] +pub(crate) struct RuntimeRequest { + pub enabled: bool, +} + +// ---------------------------------------------------------------- assembly + +fn runtime_view() -> RuntimeView { + let st = crate::plugins::runtime_status(); + RuntimeView { + installed: st.installed, + enabled: st.enabled, + running: st.running, + unit: st.unit.to_string(), + principal: st.principal, + detail: (!st.detail.is_empty()).then_some(st.detail), + } +} + +/// Build the merged catalog view: every source's shelf, annotated with what this host has and can +/// run. `force` refreshes past the TTL. +fn build_catalog(force: bool) -> CatalogResponse { + let states = store::catalogs(force); + let installed = store::installed_packages(&store::plugins_dir()); + let mut plugins = Vec::new(); + let mut source_views = Vec::new(); + + for st in &states { + let count = st.index.as_ref().map(|i| i.plugins.len()).unwrap_or(0); + source_views.push(SourceView { + name: st.source.name.clone(), + url: st.source.url.clone(), + builtin: st.source.is_official(), + signed: st.source.is_signed(), + stale: st.stale, + fetched_at: st.fetched_at, + error: st.error.clone(), + entry_count: count as u32, + public_key: st.source.public_key.clone(), + }); + let Some(idx) = &st.index else { continue }; + let verified = st.source.is_official(); + for e in &idx.plugins { + let installed_version = installed + .iter() + .find(|p| p.pkg == e.pkg) + .and_then(|p| p.version.clone()); + let reason = e.incompatible_reason(); + plugins.push(CatalogEntry { + id: e.id.clone(), + pkg: e.pkg.clone(), + title: e.title.clone(), + description: e.description.clone(), + icon: e.icon.clone(), + author: e.author.clone(), + homepage: e.homepage.clone(), + license: e.license.clone(), + version: e.version.clone(), + source: st.source.name.clone(), + // The badge is the built-in source's alone: a third-party curator can pin and + // publish, but cannot confer unom's review (D6). + tier: if verified { "verified" } else { "external" }.to_string(), + reviewed_at: verified + .then(|| e.verification.as_ref().map(|v| v.reviewed_at.clone())) + .flatten(), + platforms: e.platforms.clone(), + min_host: e.min_host.clone(), + compatible: reason.is_none(), + incompatible_reason: reason, + update_available: installed_version.as_deref().is_some_and(|v| v != e.version), + installed_version, + blocked: store::advisory_for(&e.pkg, Some(&e.version)).map(|a| a.reason), + }); + } + } + // Stable, useful order: alphabetical by title, then by source so duplicates across shelves + // don't jitter between polls. + plugins.sort_by(|a, b| { + a.title + .to_lowercase() + .cmp(&b.title.to_lowercase()) + .then_with(|| a.source.cmp(&b.source)) + }); + CatalogResponse { + host: HostFacts { + version: index::host_version().to_string(), + platform: index::HOST_PLATFORM.to_string(), + }, + sources: source_views, + plugins, + busy: jobs::busy(), + } +} + +fn build_installed(live: &[String]) -> Vec { + let dir = store::plugins_dir(); + let records = manifest::load(&dir); + let catalogs = store::cached_catalogs(); + let mut out: Vec = store::installed_packages(&dir) + .into_iter() + .map(|p| { + let rec = records.get(&p.pkg); + // The catalog row for this package, if any shelf carries it — the source of "an update + // is available" and of a friendly title for CLI-installed packages. + let entry = catalogs.iter().find_map(|s| { + s.index + .as_ref()? + .plugins + .iter() + .find(|e| e.pkg == p.pkg) + .map(|e| (e, s.source.name.clone())) + }); + let plugin_id = entry + .as_ref() + .map(|(e, _)| e.id.clone()) + .or_else(|| store::plugin_id_for_pkg(&p.pkg)); + InstalledView { + // Absence of a manifest record is meaningful: the CLI put it there (§4.5). + tier: rec + .map(|r| r.tier) + .unwrap_or(manifest::Tier::Cli) + .as_str() + .to_string(), + source: rec + .and_then(|r| r.source.clone()) + .or_else(|| entry.as_ref().map(|(_, s)| s.clone())), + entry_id: rec + .and_then(|r| r.entry_id.clone()) + .or_else(|| entry.as_ref().map(|(e, _)| e.id.clone())), + title: entry.as_ref().map(|(e, _)| e.title.clone()), + installed_at: rec.and_then(|r| r.installed_at.clone()), + running: plugin_id + .as_deref() + .is_some_and(|id| live.iter().any(|l| l == id)), + update_available: entry.as_ref().and_then(|(e, _)| { + (p.version.as_deref() != Some(e.version.as_str())).then(|| e.version.clone()) + }), + blocked: store::advisory_for(&p.pkg, p.version.as_deref()).map(|a| a.reason), + plugin_id, + version: p.version, + pkg: p.pkg, + } + }) + .collect(); + out.sort_by(|a, b| a.pkg.cmp(&b.pkg)); + out +} + +// ---------------------------------------------------------------- handlers + +/// Browse the plugin catalog +/// +/// The merged shelf across every configured source, annotated with what this host already has and +/// what it can run. Sources past their freshness window are refreshed first; a source that can't be +/// reached keeps serving its last good copy, marked `stale` (a LAN-only host still has a working +/// store — an entry's pin travelled with the entry). +#[utoipa::path( + get, + path = "/store/catalog", + tag = "store", + operation_id = "getPluginCatalog", + responses( + (status = OK, description = "The merged catalog", body = CatalogResponse), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn get_catalog() -> Response { + match blocking(|| build_catalog(false)).await { + Ok(c) => Json(c).into_response(), + Err(e) => e, + } +} + +/// Refresh every catalog now +/// +/// Bypasses the freshness window and re-fetches all sources, then returns the merged catalog. +#[utoipa::path( + post, + path = "/store/refresh", + tag = "store", + operation_id = "refreshPluginCatalog", + responses( + (status = OK, description = "The freshly-fetched catalog", body = CatalogResponse), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn refresh_catalog() -> Response { + match blocking(|| build_catalog(true)).await { + Ok(c) => { + crate::events::emit(crate::events::EventKind::StoreChanged); + Json(c).into_response() + } + Err(e) => e, + } +} + +/// List installed plugins +/// +/// What's actually in the plugins directory, joined with how it got there (the provenance manifest) +/// and whether it is registered right now. A package with no provenance record was installed with +/// the CLI and reports `tier: "cli"` — absence is the answer, not a gap. +#[utoipa::path( + get, + path = "/store/installed", + tag = "store", + operation_id = "listInstalledPlugins", + responses( + (status = OK, description = "Installed plugin packages", body = [InstalledView]), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn list_installed() -> Response { + // The live set comes from the in-memory lease registry — cheap, and it must be read on this + // thread rather than inside the blocking closure to keep the borrow simple. + let live: Vec = super::plugins::live_plugin_ids(); + match blocking(move || build_installed(&live)).await { + Ok(v) => Json(v).into_response(), + Err(e) => e, + } +} + +/// Install a plugin +/// +/// Either `{source, id}` — a catalogued entry, installed at its pinned version after its integrity +/// is re-checked against the registry — or `{spec, accept_unverified: true}`, which installs an +/// unreviewed package the operator takes responsibility for. Returns `202` with a job id; watch it +/// at `GET /store/jobs/{id}`. +/// +/// One package operation runs at a time (`409` otherwise): `bun` operations share a lockfile and a +/// `node_modules` tree. +#[utoipa::path( + post, + path = "/store/install", + tag = "store", + operation_id = "installPlugin", + request_body = InstallRequest, + responses( + (status = ACCEPTED, description = "Install job started", body = JobRef), + (status = BAD_REQUEST, description = "Unknown entry, bad spec, or missing acknowledgement", body = ApiError), + (status = CONFLICT, description = "Another package operation is in flight", body = ApiError), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn install_plugin(ApiJson(req): ApiJson) -> Response { + let plan = + match ( + req.source.as_deref(), + req.id.as_deref(), + req.spec.as_deref(), + ) { + (Some(source), Some(id), None) => { + let found = match blocking({ + let (source, id) = (source.to_string(), id.to_string()); + move || store::find_entry(&source, &id) + }) + .await + { + Ok(f) => f, + Err(e) => return e, + }; + let Some((entry, verified)) = found else { + return api_error( + StatusCode::BAD_REQUEST, + "no such plugin in that source's catalog", + ); + }; + if let Some(reason) = entry.incompatible_reason() { + return api_error( + StatusCode::BAD_REQUEST, + &format!("this plugin cannot run on this host: {reason}"), + ); + } + match jobs::Plan::from_entry(&entry, source, verified) { + Ok(p) => p, + Err(e) => return api_error(StatusCode::BAD_REQUEST, &format!("{e:#}")), + } + } + (None, None, Some(spec)) => { + if !req.accept_unverified { + return api_error( + StatusCode::BAD_REQUEST, + "installing from a raw package spec runs unreviewed code with operator \ + privileges — set `accept_unverified` to confirm", + ); + } + match jobs::Plan::from_spec(spec) { + Ok(p) => p, + Err(e) => return api_error(StatusCode::BAD_REQUEST, &format!("{e:#}")), + } + } + _ => return api_error( + StatusCode::BAD_REQUEST, + "provide either {source, id} for a catalogued plugin or {spec} for a raw package", + ), + }; + + match jobs::spawn_install(plan) { + Ok(job) => (StatusCode::ACCEPTED, Json(JobRef { job })).into_response(), + Err(e) => api_error(StatusCode::CONFLICT, &format!("{e:#}")), + } +} + +/// Uninstall a plugin +/// +/// Removes the package and forgets its provenance, then restarts the runner. Only names the runner +/// would actually supervise are accepted, so this can't be used to rip a shared dependency out of +/// the tree. +#[utoipa::path( + post, + path = "/store/uninstall", + tag = "store", + operation_id = "uninstallPlugin", + request_body = UninstallRequest, + responses( + (status = ACCEPTED, description = "Uninstall job started", body = JobRef), + (status = BAD_REQUEST, description = "Not a plugin package name", body = ApiError), + (status = CONFLICT, description = "Another package operation is in flight", body = ApiError), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn uninstall_plugin(ApiJson(req): ApiJson) -> Response { + if let Err(e) = store::valid_installed_pkg(&req.pkg) { + return api_error(StatusCode::BAD_REQUEST, &format!("{e:#}")); + } + match jobs::spawn_uninstall(req.pkg) { + Ok(job) => (StatusCode::ACCEPTED, Json(JobRef { job })).into_response(), + Err(e) => api_error(StatusCode::CONFLICT, &format!("{e:#}")), + } +} + +/// List recent package jobs +#[utoipa::path( + get, + path = "/store/jobs", + tag = "store", + operation_id = "listPluginJobs", + responses( + (status = OK, description = "Recent install/uninstall jobs, oldest first", body = [Job]), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn list_jobs() -> Json> { + Json(jobs::list()) +} + +/// Follow one package job +/// +/// Poll this while `state` is `running`; `log` carries the tail of the package manager's output. +#[utoipa::path( + get, + path = "/store/jobs/{id}", + tag = "store", + operation_id = "getPluginJob", + params(("id" = String, Path, description = "The job id returned by install/uninstall")), + responses( + (status = OK, description = "The job", body = Job), + (status = NOT_FOUND, description = "No such job (they are kept for a bounded history)", body = ApiError), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn get_job(Path(id): Path) -> Response { + match jobs::get(&id) { + Some(j) => Json(j).into_response(), + None => api_error(StatusCode::NOT_FOUND, "no such job"), + } +} + +/// List catalog sources +#[utoipa::path( + get, + path = "/store/sources", + tag = "store", + operation_id = "listPluginSources", + responses( + (status = OK, description = "Configured sources, built-in first", body = [SourceView]), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn list_sources() -> Response { + match blocking(|| { + store::cached_catalogs() + .into_iter() + .map(|st| SourceView { + name: st.source.name.clone(), + url: st.source.url.clone(), + builtin: st.source.is_official(), + signed: st.source.is_signed(), + stale: st.stale, + fetched_at: st.fetched_at, + error: st.error, + entry_count: st.index.map(|i| i.plugins.len()).unwrap_or(0) as u32, + public_key: st.source.public_key, + }) + .collect::>() + }) + .await + { + Ok(v) => Json(v).into_response(), + Err(e) => e, + } +} + +/// Add or update a catalog source +/// +/// Adding a source is a trust decision: its entries become installable on this host. They are +/// attributed to it in the console and never carry the "verified" badge, which belongs to the +/// built-in source alone. +#[utoipa::path( + put, + path = "/store/sources/{name}", + tag = "store", + operation_id = "putPluginSource", + params(("name" = String, Path, description = "Source slug (`[a-z][a-z0-9-]*`)")), + request_body = SourceInput, + responses( + (status = NO_CONTENT, description = "Source saved"), + (status = BAD_REQUEST, description = "Invalid name, url or key — or the reserved built-in name", body = ApiError), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn put_source( + Path(name): Path, + ApiJson(input): ApiJson, +) -> Response { + let source = sources::Source { + name: name.clone(), + url: input.url, + public_key: input.public_key.filter(|k| !k.trim().is_empty()), + }; + let saved = blocking(move || { + let r = sources::put(source); + if r.is_ok() { + // A redefined source must not keep serving what the old definition cached. + store::drop_source_cache(&name); + } + r + }) + .await; + match saved { + Err(e) => e, + Ok(Err(e)) => api_error(StatusCode::BAD_REQUEST, &format!("{e:#}")), + Ok(Ok(())) => { + crate::events::emit(crate::events::EventKind::StoreChanged); + StatusCode::NO_CONTENT.into_response() + } + } +} + +/// Remove a catalog source +#[utoipa::path( + delete, + path = "/store/sources/{name}", + tag = "store", + operation_id = "deletePluginSource", + params(("name" = String, Path, description = "Source slug")), + responses( + (status = NO_CONTENT, description = "Removed (or already absent)"), + (status = FORBIDDEN, description = "The built-in source cannot be removed, or the plugin token is not authorized", body = ApiError), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + ) +)] +pub(crate) async fn delete_source(Path(name): Path) -> Response { + if name == sources::OFFICIAL_NAME { + return api_error( + StatusCode::FORBIDDEN, + "the built-in source cannot be removed", + ); + } + let removed = blocking(move || { + let r = sources::remove(&name); + if matches!(r, Ok(true)) { + store::drop_source_cache(&name); + } + r + }) + .await; + match removed { + Err(e) => e, + Ok(Err(e)) => api_error(StatusCode::BAD_REQUEST, &format!("{e:#}")), + Ok(Ok(_)) => { + crate::events::emit(crate::events::EventKind::StoreChanged); + StatusCode::NO_CONTENT.into_response() + } + } +} + +/// Plugin runner state +/// +/// Installed plugins only run while the runner is on, and the runner discovers units at startup — +/// so this is both the "is anything running" answer and the explanation for a freshly installed +/// plugin that hasn't appeared yet. +#[utoipa::path( + get, + path = "/store/runtime", + tag = "store", + operation_id = "getPluginRuntime", + responses( + (status = OK, description = "Runner state", body = RuntimeView), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn get_runtime() -> Response { + match blocking(runtime_view).await { + Ok(v) => Json(v).into_response(), + Err(e) => e, + } +} + +/// Turn the plugin runner on or off +#[utoipa::path( + post, + path = "/store/runtime", + tag = "store", + operation_id = "setPluginRuntime", + request_body = RuntimeRequest, + responses( + (status = OK, description = "The resulting runner state", body = RuntimeView), + (status = BAD_REQUEST, description = "The runner could not be switched", body = ApiError), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = FORBIDDEN, description = "Not authorized for the plugin token", body = ApiError), + ) +)] +pub(crate) async fn set_runtime(ApiJson(req): ApiJson) -> Response { + let switched = + blocking(move || crate::plugins::set_runtime_enabled(req.enabled).map(|()| runtime_view())) + .await; + match switched { + Err(e) => e, + Ok(Err(e)) => api_error(StatusCode::BAD_REQUEST, &format!("{e:#}")), + Ok(Ok(v)) => { + crate::events::emit(crate::events::EventKind::StoreChanged); + Json(v).into_response() + } + } +} + +// Re-exported so `routes!` can name the response body types. +pub(crate) use crate::store::jobs::Job; diff --git a/crates/punktfunk-host/src/plugins.rs b/crates/punktfunk-host/src/plugins.rs index c0940497..a37a3acd 100644 --- a/crates/punktfunk-host/src/plugins.rs +++ b/crates/punktfunk-host/src/plugins.rs @@ -116,7 +116,11 @@ fn forward_to_runner(args: &[String]) -> Result<()> { /// Resolve how to invoke the runner CLI: the program plus any leading args (the bundled bun needs /// the runner script path passed to it). -fn runner_command() -> Result<(std::path::PathBuf, Vec)> { +/// +/// Also the plugin store's executor seam ([`crate::store::jobs`]): a console-triggered install runs +/// the *same* package ops through the *same* runner as the CLI, so there is exactly one +/// implementation of "install a plugin" on the box (design D4). +pub(crate) fn runner_command() -> Result<(std::path::PathBuf, Vec)> { #[cfg(target_os = "windows")] { // The installer lays the payload out as {app}\punktfunk-host.exe, {app}\bun\bun.exe and @@ -177,17 +181,163 @@ fn disable() -> Result<()> { Ok(()) } -#[cfg(target_os = "linux")] fn status() -> Result<()> { - let active = systemctl_output(&["is-active", UNIT]).unwrap_or_else(|| "unknown".into()); - let enabled = systemctl_output(&["is-enabled", UNIT]).unwrap_or_else(|| "unknown".into()); - println!("runner: {UNIT}\nenabled: {enabled}\nactive: {active}"); - if active != "active" { + let st = runtime_status(); + println!( + "runner: {}\nstate: {}\nenabled: {}", + st.unit, + if !st.installed { + "not installed" + } else if st.running { + "running" + } else { + "stopped" + }, + st.enabled + ); + if let Some(principal) = &st.principal { + println!("runs as: {principal}"); + } + if st.installed && !st.running { println!("\nStart it with: punktfunk-host plugins enable"); + } else if !st.installed { + println!("\n{}", st.detail); } Ok(()) } +// ---- runtime state, shared by the CLI and the plugin store's mgmt API -------------------------- + +/// Whether the plugin runner is present, switched on, and up. +/// +/// The store's console surface needs this as data (to offer "enable the runner" before the first +/// install, and to explain why a freshly installed plugin isn't running yet), so it lives here +/// rather than being formatted straight to stdout like the CLI once did. +#[derive(Debug, Clone)] +pub(crate) struct RuntimeStatus { + /// Is the runner payload / service unit on this box at all? + pub installed: bool, + /// Is it configured to start (systemd `enabled`, or a non-`Disabled` scheduled task)? + pub enabled: bool, + /// Is it up right now? + pub running: bool, + /// The unit / task name, so operator-facing copy can name the thing to look at. + pub unit: &'static str, + /// Windows: the account the task runs as (the SYSTEM→LocalService migration is visible here). + pub principal: Option, + /// One line of human-readable context, mostly for the "not installed" case. + pub detail: String, +} + +#[cfg(target_os = "linux")] +pub(crate) fn runtime_status() -> RuntimeStatus { + let enabled_raw = systemctl_output(&["is-enabled", UNIT]); + let active = systemctl_output(&["is-active", UNIT]).unwrap_or_default(); + // `is-enabled` answers `not-found` when the unit file isn't installed at all; the runner + // payload being present is the other half of "can we install plugins". + let unit_known = enabled_raw.as_deref().is_some_and(|s| s != "not-found"); + let installed = unit_known || runner_command().is_ok(); + RuntimeStatus { + installed, + enabled: enabled_raw.as_deref() == Some("enabled"), + running: active == "active", + unit: UNIT, + principal: None, + detail: if installed { + String::new() + } else { + "the plugin runner package isn't installed (Debian/Ubuntu: `sudo apt install \ + punktfunk-scripting`)" + .into() + }, + } +} + +#[cfg(target_os = "windows")] +pub(crate) fn runtime_status() -> RuntimeStatus { + let out = powershell_output(&format!( + "$t = Get-ScheduledTask -TaskName {TASK} -ErrorAction SilentlyContinue; \ + if ($null -eq $t) {{ 'missing' }} else {{ \"$($t.State)|$($t.Principal.UserId)\" }}" + )); + match out.as_deref().map(str::trim) { + Some("missing") | None => RuntimeStatus { + installed: false, + enabled: false, + running: false, + unit: TASK, + principal: None, + detail: "reinstall punktfunk with the scripting component to get the plugin runner" + .into(), + }, + Some(raw) => { + let (state, principal) = raw.split_once('|').unwrap_or((raw, "")); + RuntimeStatus { + installed: true, + enabled: !state.eq_ignore_ascii_case("Disabled"), + running: state.eq_ignore_ascii_case("Running"), + unit: TASK, + principal: (!principal.is_empty()).then(|| principal.to_string()), + detail: String::new(), + } + } + } +} + +#[cfg(not(any(target_os = "linux", target_os = "windows")))] +pub(crate) fn runtime_status() -> RuntimeStatus { + RuntimeStatus { + installed: false, + enabled: false, + running: false, + unit: "punktfunk-scripting", + principal: None, + detail: "the plugin runner is only available on Linux and Windows hosts".into(), + } +} + +/// Switch the runner on or off — the [`enable`]/[`disable`] the CLI runs, exposed for the store's +/// `POST /store/runtime`. On Windows this is reached from the SYSTEM service, which already clears +/// the elevation bar the CLI has to check for. +pub(crate) fn set_runtime_enabled(enabled: bool) -> Result<()> { + if enabled { + enable() + } else { + disable() + } +} + +/// Restart the runner so it rediscovers installed units. Returns `false` (not an error) when it +/// isn't running — there is nothing to restart, and the store reports that as "installed, but the +/// runner is off" rather than as a failure. +/// +/// Unit discovery happens once at runner startup ([`sdk/src/runner.ts`]), so this restart *is* the +/// activation step for a newly installed plugin. +pub(crate) fn restart_runtime() -> Result { + let st = runtime_status(); + if !st.installed || !st.running { + return Ok(false); + } + #[cfg(target_os = "linux")] + { + run_systemctl(&["restart", UNIT])?; + Ok(true) + } + #[cfg(target_os = "windows")] + { + // Stop then start: `Restart-ScheduledTask` does not exist, and a Start on an already- + // running task is a no-op rather than a restart. + powershell(&format!( + "Stop-ScheduledTask -TaskName {TASK} -ErrorAction SilentlyContinue; \ + Start-ScheduledTask -TaskName {TASK} -ErrorAction Stop" + ))?; + Ok(true) + } + #[cfg(not(any(target_os = "linux", target_os = "windows")))] + { + Ok(false) + } +} + #[cfg(target_os = "linux")] fn run_systemctl(args: &[&str]) -> Result<()> { let status = Command::new("systemctl") @@ -451,33 +601,6 @@ fn icacls_path() -> String { .unwrap_or_else(|_| "icacls".to_string()) } -#[cfg(target_os = "windows")] -fn status() -> Result<()> { - let out = powershell_output(&format!( - "$t = Get-ScheduledTask -TaskName {TASK} -ErrorAction SilentlyContinue; \ - if ($null -eq $t) {{ 'missing' }} else {{ \ - \"$($t.State)|$($t.Principal.UserId)\" }}" - )); - match out.as_deref().map(str::trim) { - Some("missing") | None => { - println!( - "runner: {TASK}\nstate: not installed\n\nReinstall punktfunk with the \ - scripting component to get the plugin runner." - ); - } - Some(state) => { - // "State|Principal" — the principal line makes the SYSTEM→LocalService migration - // verifiable at a glance (`plugins enable` converges a legacy SYSTEM task). - let (state, principal) = state.split_once('|').unwrap_or((state, "?")); - println!("runner: {TASK}\nstate: {state}\nruns as: {principal}"); - if state.eq_ignore_ascii_case("Disabled") { - println!("\nEnable it with: punktfunk-host plugins enable"); - } - } - } - Ok(()) -} - /// Resolve powershell by full System32 path rather than PATH — CreateProcess searches the launching /// EXE's own directory first, so a planted `powershell.exe` beside the host binary would otherwise /// run with our privileges (security-review 2026-07-17; matches service.rs / pf_vdisplay). @@ -603,8 +726,3 @@ fn enable() -> Result<()> { fn disable() -> Result<()> { bail!("the plugin runner is only available on Linux and Windows hosts") } - -#[cfg(not(any(target_os = "linux", target_os = "windows")))] -fn status() -> Result<()> { - bail!("the plugin runner is only available on Linux and Windows hosts") -} diff --git a/crates/punktfunk-host/src/store.rs b/crates/punktfunk-host/src/store.rs new file mode 100644 index 00000000..89cccae6 --- /dev/null +++ b/crates/punktfunk-host/src/store.rs @@ -0,0 +1,394 @@ +//! The **plugin store**: discovering and installing plugins from signed catalogs +//! (design `plugin-store.md`). +//! +//! Installing a plugin means running somebody's code with operator privileges, so the store is +//! built around one idea: *the index is the verification gate*. A catalog entry pins one exact +//! version and that version's tarball hash, and nothing here can express "track the latest +//! release". Upstream can publish whatever they like; this host will keep offering the reviewed +//! version until a newly reviewed entry lands in a signed index. +//! +//! That yields three tiers, which the whole surface is organized around: +//! +//! | tier | where it came from | surfaced? | +//! |------|--------------------|-----------| +//! | **verified** | the built-in `unom` source's index — unom reviewed this exact tarball | yes, with a badge | +//! | **external** | an operator-added source's index — pinned and integrity-checked, curated by somebody else | yes, attributed, no badge | +//! | **unverified** | a raw package spec typed into the console's danger dialog | never listed; install only | +//! +//! This module is the domain half (catalog state, installed-package facts, trust decisions); the +//! HTTP surface lives in [`crate::mgmt::store`]. Everything here is **blocking** — callers on the +//! async side hand it to `spawn_blocking`. + +pub(crate) mod catalog; +pub(crate) mod index; +pub(crate) mod jobs; +pub(crate) mod manifest; +pub(crate) mod sources; + +use anyhow::{bail, Result}; +use index::{Advisory, Entry, Index}; +use sources::Source; +use std::path::{Path, PathBuf}; +use std::sync::RwLock; + +/// How long a fetched catalog is considered fresh. Catalogs change when somebody reviews a +/// release — hours, not seconds — and a streaming host should not be chatting to the internet on a +/// timer for a page nobody has open. +const CATALOG_TTL_SECS: u64 = 6 * 60 * 60; + +/// Where plugin packages live: `/plugins` (the same dir the SDK and runner use). +pub(crate) fn plugins_dir() -> PathBuf { + pf_paths::config_dir().join("plugins") +} + +// ---------------------------------------------------------------- installed packages + +/// A plugin package present in the plugins dir. +#[derive(Debug, Clone)] +pub(crate) struct InstalledPkg { + pub pkg: String, + pub version: Option, +} + +/// Enumerate installed plugin packages under `/node_modules`. +/// +/// Mirrors the runner's own discovery so the store never claims something is installed that the +/// runner wouldn't supervise: the unscoped `punktfunk-plugin-*` convention, and **any** scope's +/// `plugin-*` (`@punktfunk/plugin-rom-manager`, `@retro-hub/plugin-x`). Scoped-any is what makes a +/// third-party catalog entry work at all — a scoped name is required for the registry mapping +/// (D8), so discovery must not be limited to the first-party scope. +pub(crate) fn installed_packages(dir: &Path) -> Vec { + let modules = dir.join("node_modules"); + let mut out = Vec::new(); + let version_of = |pkg_dir: &Path| -> Option { + let bytes = std::fs::read(pkg_dir.join("package.json")).ok()?; + let v: serde_json::Value = serde_json::from_slice(&bytes).ok()?; + v.get("version")?.as_str().map(str::to_string) + }; + let Ok(entries) = std::fs::read_dir(&modules) else { + return out; // nothing installed yet + }; + let mut names: Vec = entries + .filter_map(|e| e.ok()) + .map(|e| e.file_name().to_string_lossy().into_owned()) + .collect(); + names.sort(); + for name in names { + if name.starts_with("punktfunk-plugin-") { + let dir = modules.join(&name); + out.push(InstalledPkg { + version: version_of(&dir), + pkg: name, + }); + } else if name.starts_with('@') { + let scope_dir = modules.join(&name); + let Ok(scoped) = std::fs::read_dir(&scope_dir) else { + continue; + }; + let mut inner: Vec = scoped + .filter_map(|e| e.ok()) + .map(|e| e.file_name().to_string_lossy().into_owned()) + .collect(); + inner.sort(); + for s in inner { + if s.starts_with("plugin-") { + let dir = scope_dir.join(&s); + out.push(InstalledPkg { + pkg: format!("{name}/{s}"), + version: version_of(&dir), + }); + } + } + } + } + out +} + +/// Is `pkg` a name the runner would supervise? Guards the uninstall route so a stray +/// `POST /store/uninstall {"pkg": "effect"}` can't rip a shared dependency out of the tree. +pub(crate) fn valid_installed_pkg(pkg: &str) -> Result<()> { + let plausible = pkg.starts_with("punktfunk-plugin-") + || (index::valid_scoped_pkg(pkg) + && pkg + .split_once('/') + .is_some_and(|(_, name)| name.starts_with("plugin-"))); + if !plausible { + bail!("`{pkg}` is not a plugin package (`@scope/plugin-*` or `punktfunk-plugin-*`)"); + } + Ok(()) +} + +/// The plugin id a package registers under (`@punktfunk/plugin-rom-manager` → `rom-manager`), used +/// to line an installed package up with the live lease registry. Best-effort: the authoritative id +/// for a catalogued plugin is its index entry. +pub(crate) fn plugin_id_for_pkg(pkg: &str) -> Option { + let last = pkg.rsplit('/').next()?; + let id = last + .strip_prefix("punktfunk-plugin-") + .or_else(|| last.strip_prefix("plugin-"))?; + index::valid_plugin_id(id).then(|| id.to_string()) +} + +// ---------------------------------------------------------------- catalog state + +/// What we hold for one source: the last good index plus why it might be stale. +#[derive(Clone)] +pub(crate) struct SourceState { + pub source: Source, + pub index: Option, + /// Unix seconds of the fetch that produced [`Self::index`]. + pub fetched_at: Option, + /// True when the last refresh attempt failed and we're serving an older copy. + pub stale: bool, + /// Why the last attempt failed, for the console to show against the source. + pub error: Option, + pub etag: Option, +} + +impl SourceState { + fn empty(source: Source) -> Self { + Self { + source, + index: None, + fetched_at: None, + stale: false, + error: None, + etag: None, + } + } + + fn is_fresh(&self) -> bool { + self.index.is_some() + && self + .fetched_at + .is_some_and(|t| catalog::unix_now().saturating_sub(t) < CATALOG_TTL_SECS) + } +} + +fn state() -> &'static RwLock> { + static STATE: std::sync::OnceLock>> = std::sync::OnceLock::new(); + STATE.get_or_init(|| RwLock::new(Vec::new())) +} + +/// Reconcile the in-memory state list with the configured sources (added/removed/edited), seeding +/// anything new from the on-disk cache so a cold host has a browsable store before its first fetch. +fn sync_sources() { + let configured = sources::load(); + let dir = catalog::cache_dir(); + let mut st = state().write().unwrap_or_else(|e| e.into_inner()); + st.retain(|s| configured.iter().any(|c| c.name == s.source.name)); + for c in configured { + match st.iter_mut().find(|s| s.source.name == c.name) { + // The URL or key may have been edited under a name we already hold — drop what we + // cached for the old definition rather than attributing it to the new one. + Some(existing) => { + if existing.source.url != c.url || existing.source.public_key != c.public_key { + *existing = SourceState::empty(c); + } else { + existing.source = c; + } + } + None => { + let mut fresh = SourceState::empty(c.clone()); + if let Some((index, meta)) = catalog::read_cache(&dir, &c.name) { + fresh.index = Some(index); + fresh.fetched_at = Some(meta.fetched_at); + fresh.etag = meta.etag; + fresh.stale = true; // from disk — unverified freshness until a fetch says otherwise + } + st.push(fresh); + } + } + } +} + +/// Every source's catalog, refreshing those past their TTL (or all of them when `force`). +/// +/// **Blocking** — network I/O. The freshness decision is ours (our fetch clock), never the +/// document's self-reported `generated` field. +pub(crate) fn catalogs(force: bool) -> Vec { + sync_sources(); + let dir = catalog::cache_dir(); + let todo: Vec = { + let st = state().read().unwrap_or_else(|e| e.into_inner()); + st.iter() + .filter(|s| force || !s.is_fresh()) + .map(|s| s.source.clone()) + .collect() + }; + for source in todo { + let etag = { + let st = state().read().unwrap_or_else(|e| e.into_inner()); + st.iter() + .find(|s| s.source.name == source.name) + .and_then(|s| s.etag.clone()) + }; + let outcome = catalog::fetch(&source, etag.as_deref()); + let now = catalog::unix_now(); + let mut st = state().write().unwrap_or_else(|e| e.into_inner()); + let Some(slot) = st.iter_mut().find(|s| s.source.name == source.name) else { + continue; // removed while we were fetching + }; + match outcome { + catalog::Fetched::Fresh { index, etag } => { + let count = index.plugins.len(); + catalog::write_cache( + &dir, + &source.name, + &index, + &catalog::CacheMeta { + etag: etag.clone(), + fetched_at: now, + }, + ); + slot.index = Some(*index); + slot.fetched_at = Some(now); + slot.etag = etag; + slot.stale = false; + slot.error = None; + tracing::info!(source = %source.name, entries = count, "plugin catalog refreshed"); + } + catalog::Fetched::NotModified => { + slot.fetched_at = Some(now); + slot.stale = false; + slot.error = None; + } + catalog::Fetched::Failed(why) => { + // Fail soft: keep the last good shelf, say plainly that it's stale. + tracing::warn!(source = %source.name, "plugin catalog refresh failed: {why}"); + slot.stale = true; + slot.error = Some(why); + } + } + } + state() + .read() + .unwrap_or_else(|e| e.into_inner()) + .iter() + .cloned() + .collect() +} + +/// The catalogs we already hold, without touching the network. For paths that must not block on a +/// remote host (an install resolving its own entry, an advisory lookup). +pub(crate) fn cached_catalogs() -> Vec { + sync_sources(); + state() + .read() + .unwrap_or_else(|e| e.into_inner()) + .iter() + .cloned() + .collect() +} + +/// Find a catalog entry by `(source, id)`, plus whether that source is the built-in one — which is +/// the single place "verified" is decided (D6). +pub(crate) fn find_entry(source_name: &str, id: &str) -> Option<(Entry, bool)> { + cached_catalogs().into_iter().find_map(|s| { + if s.source.name != source_name { + return None; + } + let entry = s.index?.plugins.into_iter().find(|e| e.id == id)?; + Some((entry, s.source.is_official())) + }) +} + +/// The first advisory covering `pkg@version`, across every source we hold. Revocations are +/// deliberately not scoped to the source a plugin came from: a package known-bad anywhere is +/// known-bad here. +pub(crate) fn advisory_for(pkg: &str, version: Option<&str>) -> Option { + let version = version?; + cached_catalogs().into_iter().find_map(|s| { + s.index? + .security + .into_iter() + .find(|a| a.matches(pkg, version)) + }) +} + +/// Forget a removed source's cached shelf, so re-adding the name later can't serve stale rows. +pub(crate) fn drop_source_cache(name: &str) { + catalog::drop_cache(&catalog::cache_dir(), name); + let mut st = state().write().unwrap_or_else(|e| e.into_inner()); + st.retain(|s| s.source.name != name); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn touch_pkg(root: &Path, pkg: &str, version: &str) { + let dir = root.join("node_modules").join(pkg); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write( + dir.join("package.json"), + format!(r#"{{"name":"{pkg}","version":"{version}"}}"#), + ) + .unwrap(); + } + + #[test] + fn scans_both_conventions_and_any_scope() { + let dir = tempfile::tempdir().unwrap(); + touch_pkg(dir.path(), "@punktfunk/plugin-rom-manager", "0.3.0"); + // A third-party scoped plugin MUST be found: catalog entries are required to be scoped + // (D8), so limiting discovery to @punktfunk would make every external source unusable. + touch_pkg(dir.path(), "@retro-hub/plugin-x", "1.0.0"); + touch_pkg(dir.path(), "punktfunk-plugin-legacy", "0.1.0"); + // Not plugins: a plain dependency and a scoped non-plugin. + touch_pkg(dir.path(), "effect", "4.0.0"); + touch_pkg(dir.path(), "@punktfunk/host", "0.1.2"); + + let found = installed_packages(dir.path()); + let names: Vec<&str> = found.iter().map(|p| p.pkg.as_str()).collect(); + assert!( + names.contains(&"@punktfunk/plugin-rom-manager"), + "{names:?}" + ); + assert!(names.contains(&"@retro-hub/plugin-x"), "{names:?}"); + assert!(names.contains(&"punktfunk-plugin-legacy"), "{names:?}"); + assert!(!names.contains(&"effect"), "{names:?}"); + assert!(!names.contains(&"@punktfunk/host"), "{names:?}"); + assert_eq!( + found + .iter() + .find(|p| p.pkg == "@punktfunk/plugin-rom-manager") + .unwrap() + .version + .as_deref(), + Some("0.3.0") + ); + } + + #[test] + fn scan_of_a_missing_dir_is_empty_not_an_error() { + let dir = tempfile::tempdir().unwrap(); + assert!(installed_packages(dir.path()).is_empty()); + } + + #[test] + fn uninstall_target_must_be_a_plugin_package() { + assert!(valid_installed_pkg("@punktfunk/plugin-rom-manager").is_ok()); + assert!(valid_installed_pkg("@retro-hub/plugin-x").is_ok()); + assert!(valid_installed_pkg("punktfunk-plugin-legacy").is_ok()); + // A shared dependency is not removable through the store. + assert!(valid_installed_pkg("effect").is_err()); + assert!(valid_installed_pkg("@punktfunk/host").is_err()); + assert!(valid_installed_pkg("../../etc").is_err()); + assert!(valid_installed_pkg("").is_err()); + } + + #[test] + fn plugin_id_derivation() { + assert_eq!( + plugin_id_for_pkg("@punktfunk/plugin-rom-manager").as_deref(), + Some("rom-manager") + ); + assert_eq!( + plugin_id_for_pkg("punktfunk-plugin-playnite").as_deref(), + Some("playnite") + ); + assert_eq!(plugin_id_for_pkg("@a/plugin-x").as_deref(), Some("x")); + assert_eq!(plugin_id_for_pkg("effect"), None); + } +} diff --git a/crates/punktfunk-host/src/store/catalog.rs b/crates/punktfunk-host/src/store/catalog.rs new file mode 100644 index 00000000..b675c406 --- /dev/null +++ b/crates/punktfunk-host/src/store/catalog.rs @@ -0,0 +1,255 @@ +//! Catalog **fetch + cache**: pulling a source's signed index over HTTPS and keeping the last good +//! copy on disk (design `plugin-store.md` §4.1). +//! +//! Two properties matter more than freshness here: +//! +//! - **Fail closed on signatures, fail soft on the network.** A source with a pinned key whose +//! document doesn't verify is an *error* — we keep serving the last good copy and say so, and we +//! never fall back to "well, unsigned then". But a box that simply can't reach the internet +//! (LAN-only, common for a streaming host) keeps a working store: the cached shelf still browses, +//! and installs off it still pin and integrity-check, because the pin travelled with the entry. +//! - **Bounded everything.** Size cap, timeout, redirect cap, https-only, no credentials ever +//! attached. A source URL is operator config, not request-time input, so there is no lever here +//! for a browser to aim the host at an arbitrary address. + +use super::index::{Index, MAX_INDEX_BYTES}; +use super::sources::Source; +use std::io::Read; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +/// Wall-clock budget for one index (or signature) fetch. +const FETCH_TIMEOUT: Duration = Duration::from_secs(15); + +/// What a refresh attempt produced. +pub(crate) enum Fetched { + /// A new, verified, parsed document. + Fresh { + index: Box, + etag: Option, + }, + /// The source says our cached copy is still current (HTTP 304). + NotModified, + /// Nothing usable arrived. The caller keeps whatever it had and marks the source stale. + Failed(String), +} + +/// Fetch, verify and parse a source's index. +/// +/// Blocking (`ureq`) — callers run this on a blocking thread, never on the async runtime. +pub(crate) fn fetch(source: &Source, etag: Option<&str>) -> Fetched { + if !source.url.starts_with("https://") { + return Fetched::Failed("source url must be https".into()); + } + let agent = ureq::AgentBuilder::new() + .timeout(FETCH_TIMEOUT) + // A signed document doesn't need many hops to reach us; a redirect chain is a good way to + // waste a host's time. + .redirects(3) + .user_agent(&format!("punktfunk-host/{}", super::index::host_version())) + .build(); + + let mut req = agent.get(&source.url); + if let Some(tag) = etag { + req = req.set("If-None-Match", tag); + } + let resp = match req.call() { + Ok(r) => r, + // 304 is the happy "nothing changed" path, not a failure. + Err(ureq::Error::Status(304, _)) => return Fetched::NotModified, + Err(ureq::Error::Status(code, _)) => { + return Fetched::Failed(format!("index fetch returned HTTP {code}")) + } + Err(e) => return Fetched::Failed(format!("index fetch failed: {e}")), + }; + let new_etag = resp.header("etag").map(str::to_string); + let body = match read_capped(resp) { + Ok(b) => b, + Err(e) => return Fetched::Failed(e), + }; + + // Signature FIRST — nothing below may look at a field before this passes (design §6.3). + let keys = source.keys(); + if !keys.is_empty() { + let sig = match agent.get(&source.sig_url()).call() { + Ok(r) => match read_capped(r) { + Ok(b) => b, + Err(e) => return Fetched::Failed(format!("signature: {e}")), + }, + Err(e) => { + return Fetched::Failed(format!( + "this source is signed but its signature could not be fetched: {e}" + )) + } + }; + let sig_text = match String::from_utf8(sig) { + Ok(s) => s, + Err(_) => return Fetched::Failed("signature file is not text".into()), + }; + if let Err(e) = super::index::verify_signature(&body, &sig_text, &keys) { + return Fetched::Failed(format!("signature check failed: {e}")); + } + } + + match Index::parse(&body) { + Ok(index) => Fetched::Fresh { + index: Box::new(index), + etag: new_etag, + }, + Err(e) => Fetched::Failed(format!("{e:#}")), + } +} + +/// Read a response body, refusing anything past the cap without buffering it. +fn read_capped(resp: ureq::Response) -> Result, String> { + let mut buf = Vec::new(); + resp.into_reader() + .take((MAX_INDEX_BYTES + 1) as u64) + .read_to_end(&mut buf) + .map_err(|e| format!("reading the response body failed: {e}"))?; + if buf.len() > MAX_INDEX_BYTES { + return Err(format!("response exceeds the {MAX_INDEX_BYTES}-byte cap")); + } + Ok(buf) +} + +// ---------------------------------------------------------------- disk cache + +/// `/store-cache` — the last good copy of every source's index, so a host that boots +/// without a network still has a browsable store. +pub(crate) fn cache_dir() -> PathBuf { + pf_paths::config_dir().join("store-cache") +} + +fn body_path(dir: &Path, source: &str) -> PathBuf { + dir.join(format!("{source}.json")) +} + +fn meta_path(dir: &Path, source: &str) -> PathBuf { + dir.join(format!("{source}.meta.json")) +} + +#[derive(serde::Serialize, serde::Deserialize, Default)] +pub(crate) struct CacheMeta { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub etag: Option, + /// Unix seconds of the fetch that produced the cached body. + #[serde(default)] + pub fetched_at: u64, +} + +/// The cached index for a source, if one is on disk and still parses. Re-validated on read: a +/// cache file is just as untrusted as the wire (it lives in a directory an admin can write). +/// +/// NOTE the signature is **not** re-checked here — it was checked when the bytes were accepted, +/// and the cache directory is inside the host's own private config tree. +pub(crate) fn read_cache(dir: &Path, source: &str) -> Option<(Index, CacheMeta)> { + let body = std::fs::read(body_path(dir, source)).ok()?; + let index = Index::parse(&body).ok()?; + let meta = std::fs::read(meta_path(dir, source)) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()) + .unwrap_or_default(); + Some((index, meta)) +} + +/// Persist a freshly verified index as the new last-good copy. +pub(crate) fn write_cache(dir: &Path, source: &str, index: &Index, meta: &CacheMeta) { + if let Err(e) = pf_paths::create_private_dir(dir) { + tracing::warn!("could not create the store cache dir: {e}"); + return; + } + let write = |path: PathBuf, bytes: Vec| { + let tmp = path.with_extension("tmp"); + if std::fs::write(&tmp, bytes).is_ok() { + let _ = std::fs::rename(&tmp, &path); + } + }; + match serde_json::to_vec_pretty(index) { + Ok(b) => write(body_path(dir, source), b), + Err(e) => tracing::warn!("could not serialize the catalog cache: {e}"), + } + if let Ok(b) = serde_json::to_vec_pretty(meta) { + write(meta_path(dir, source), b); + } +} + +/// Drop a removed source's cache files so a later re-add can't be served a stale shelf. +pub(crate) fn drop_cache(dir: &Path, source: &str) { + let _ = std::fs::remove_file(body_path(dir, source)); + let _ = std::fs::remove_file(meta_path(dir, source)); +} + +pub(crate) fn unix_now() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sample_index() -> Index { + Index::parse( + br#"{"schema":1,"name":"t","plugins":[{"id":"a","pkg":"@p/plugin-a", + "registry":"https://r.example/","title":"A","version":"1.0.0", + "integrity":"sha512-AAAA"}]}"#, + ) + .unwrap() + } + + #[test] + fn cache_round_trips() { + let dir = tempfile::tempdir().unwrap(); + let idx = sample_index(); + let meta = CacheMeta { + etag: Some("\"abc\"".into()), + fetched_at: 1_700_000_000, + }; + write_cache(dir.path(), "unom", &idx, &meta); + + let (back, back_meta) = read_cache(dir.path(), "unom").expect("cache should be readable"); + assert_eq!(back.plugins.len(), 1); + assert_eq!(back.plugins[0].id, "a"); + assert_eq!(back_meta.etag.as_deref(), Some("\"abc\"")); + assert_eq!(back_meta.fetched_at, 1_700_000_000); + } + + #[test] + fn missing_or_corrupt_cache_reads_as_none() { + let dir = tempfile::tempdir().unwrap(); + assert!(read_cache(dir.path(), "nope").is_none()); + std::fs::write(dir.path().join("bad.json"), b"{ not json").unwrap(); + assert!(read_cache(dir.path(), "bad").is_none()); + } + + #[test] + fn cache_is_revalidated_on_read() { + // A tampered cache file (schema bumped by hand) must not be served. + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("x.json"), br#"{"schema":99,"plugins":[]}"#).unwrap(); + assert!(read_cache(dir.path(), "x").is_none()); + } + + #[test] + fn drop_cache_removes_both_files() { + let dir = tempfile::tempdir().unwrap(); + write_cache(dir.path(), "s", &sample_index(), &CacheMeta::default()); + assert!(read_cache(dir.path(), "s").is_some()); + drop_cache(dir.path(), "s"); + assert!(read_cache(dir.path(), "s").is_none()); + assert!(!meta_path(dir.path(), "s").exists()); + } + + #[test] + fn non_https_source_never_reaches_the_network() { + let s = Source { + name: "x".into(), + url: "http://example.org/i.json".into(), + public_key: None, + }; + assert!(matches!(fetch(&s, None), Fetched::Failed(_))); + } +} diff --git a/crates/punktfunk-host/src/store/index.rs b/crates/punktfunk-host/src/store/index.rs new file mode 100644 index 00000000..dfab07ab --- /dev/null +++ b/crates/punktfunk-host/src/store/index.rs @@ -0,0 +1,570 @@ +//! The plugin-store **index**: the catalog document a source serves, and the ed25519 signature +//! that makes it trustworthy (design `plugin-store.md` §3.2). +//! +//! The index is the verification gate. Each entry pins **one exact version plus that version's +//! tarball integrity hash** — so "verified on every release" is a property of the data, not a +//! promise about process: upstream can publish a newer version, but nothing in this host will +//! offer it until a re-reviewed entry lands in a signed index. There is deliberately no way to +//! express "track latest" for a catalogued plugin. +//! +//! Two rules keep parsing safe: +//! 1. **Signature before parse.** [`verify_signature`] runs over the exact bytes; only then does +//! anything here look at a field. A source with a pinned key whose signature fails is an +//! error, never a fallback to "unsigned". +//! 2. **Validate every field, drop what fails.** A malformed *entry* is dropped with a warning +//! (one bad row shouldn't blind the operator to the rest of the shelf); a malformed +//! *document* — unknown schema, non-JSON, oversized — is rejected whole. + +use anyhow::{bail, Context, Result}; +use serde::{Deserialize, Serialize}; + +/// The only index schema this host understands. A future breaking change bumps this and old +/// hosts report the source as unreadable rather than guessing at unknown semantics. +pub(crate) const SCHEMA: u32 = 1; + +/// Hard cap on a fetched index body. Generous for a text catalog (the seed index is ~2 KB) and +/// small enough that a hostile or broken source can't exhaust memory. +pub(crate) const MAX_INDEX_BYTES: usize = 5 * 1024 * 1024; + +/// Caps on how much of a (validly signed) index we'll keep. A curated shelf is small; these exist +/// so a compromised signing key can't turn the console into an unbounded list. +const MAX_PLUGINS: usize = 500; +const MAX_ADVISORIES: usize = 200; + +// ---------------------------------------------------------------- the document + +/// A source's catalog document, as served (and signed). +#[derive(Debug, Clone, Deserialize, Serialize)] +pub(crate) struct Index { + /// Document schema — must equal [`SCHEMA`]. + pub schema: u32, + /// Human-readable name of the catalog ("unom official"). Display only. + #[serde(default)] + pub name: String, + /// RFC-3339 build timestamp. Display only — freshness is decided by our own fetch clock, + /// never by a field the source controls. + #[serde(default)] + pub generated: String, + /// The shelf. + #[serde(default)] + pub plugins: Vec, + /// Revocations/advisories, checked against **installed** packages as well as catalog rows. + #[serde(default)] + pub security: Vec, +} + +/// One catalogued plugin: exactly one installable version, pinned by integrity hash. +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Entry { + /// The plugin's `definePlugin` id (`[a-z][a-z0-9-]*`) — also how a running plugin appears in + /// the lease registry, so the console can tell "catalogued" from "running". + pub id: String, + /// npm package name. **Must be scoped** (`@scope/name`): the scope is what maps the package + /// to [`Entry::registry`] in bun's `bunfig.toml`, so an unscoped name would be ambiguous + /// about where it installs from (design D8). + pub pkg: String, + /// The npm registry this package installs from. `https://` only. + pub registry: String, + pub title: String, + #[serde(default)] + pub description: String, + /// Lucide icon name for the console card (`[a-z0-9-]{1,48}`), matched against the console's + /// curated set; anything unknown falls back to a generic puzzle piece. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub icon: Option, + #[serde(default)] + pub author: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub homepage: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub license: Option, + /// **The** installable version — exact, never a range. One version per entry: the index is a + /// curated shelf, not a version archive (older releases live in the index repo's git history). + pub version: String, + /// npm integrity of that version's tarball (`sha512-`). Checked against the registry's + /// own advertised integrity before any install runs — a republish under the same version with + /// different content cannot pass. + pub integrity: String, + /// Present when a human reviewed this exact tarball. Only meaningful in the built-in source's + /// index; a third-party source can write it, which is why it never grants the "Verified" badge + /// (design D6). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub verification: Option, + /// Minimum host version this plugin supports (semver). Incompatible rows still list, greyed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub min_host: Option, + /// Host platforms this plugin works on (`linux`/`windows`/`macos`). Empty ⇒ all. + #[serde(default)] + pub platforms: Vec, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Verification { + /// ISO date the review landed. Display only. + pub reviewed_at: String, +} + +/// A revocation: "this package at these versions is known-bad". Checked against what's *installed*, +/// not just what's listed — the whole point is to reach plugins already on the box. +#[derive(Debug, Clone, Deserialize, Serialize)] +pub(crate) struct Advisory { + pub pkg: String, + /// A semver requirement (`<0.3.2`, `>=1.0.0, <1.2.0`, `*`). Unparseable ⇒ the advisory is + /// dropped rather than applied to everything. + pub versions: String, + pub reason: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub url: Option, +} + +// ---------------------------------------------------------------- signature + +/// An ed25519 public key pinned by a source record, spelled `ed25519:`. +#[derive(Debug, Clone)] +pub(crate) struct PublicKey(Vec); + +impl PublicKey { + pub(crate) fn parse(s: &str) -> Result { + use base64::Engine as _; + let b64 = s + .strip_prefix("ed25519:") + .context("public key must be spelled `ed25519:`")?; + let raw = base64::engine::general_purpose::STANDARD + .decode(b64.trim()) + .context("public key is not valid base64")?; + if raw.len() != 32 { + bail!("ed25519 public key must be 32 bytes, got {}", raw.len()); + } + Ok(Self(raw)) + } +} + +/// Verify a detached ed25519 signature over the **exact** index bytes against any of the pinned +/// keys (two slots, so a key rotation is "sign with the new one, ship a host that trusts both, +/// retire the old" rather than a flag day). +/// +/// `sig_text` is the `.sig` file's contents: base64, whitespace-tolerant. +pub(crate) fn verify_signature(bytes: &[u8], sig_text: &str, keys: &[PublicKey]) -> Result<()> { + use base64::Engine as _; + if keys.is_empty() { + bail!("no public key pinned for this source"); + } + let sig = base64::engine::general_purpose::STANDARD + .decode(sig_text.trim()) + .context("signature file is not valid base64")?; + for key in keys { + let pk = ring::signature::UnparsedPublicKey::new(&ring::signature::ED25519, &key.0); + if pk.verify(bytes, &sig).is_ok() { + return Ok(()); + } + } + bail!("index signature does not verify against any pinned key") +} + +// ---------------------------------------------------------------- parse + validate + +impl Index { + /// Parse and validate an index document. Rejects the whole document on a structural problem; + /// drops individual entries/advisories that fail validation (with a warning) so one bad row + /// can't hide the rest of the catalog. + pub(crate) fn parse(bytes: &[u8]) -> Result { + if bytes.len() > MAX_INDEX_BYTES { + bail!("index is larger than the {MAX_INDEX_BYTES}-byte cap"); + } + let mut idx: Index = serde_json::from_slice(bytes).context("index is not valid JSON")?; + if idx.schema != SCHEMA { + bail!( + "unsupported index schema {} (this host understands {SCHEMA})", + idx.schema + ); + } + idx.name = sanitize(&idx.name, 64); + idx.generated = sanitize(&idx.generated, 40); + + let mut seen: Vec = Vec::new(); + idx.plugins.truncate(MAX_PLUGINS); + idx.plugins.retain_mut(|e| match e.validate() { + Err(why) => { + tracing::warn!(pkg = %e.pkg, "dropping catalog entry: {why}"); + false + } + Ok(()) => { + // Duplicate ids would make "install this one" ambiguous; first wins. + if seen.iter().any(|s| s == &e.id) { + tracing::warn!(id = %e.id, "dropping duplicate catalog entry"); + false + } else { + seen.push(e.id.clone()); + true + } + } + }); + + idx.security.truncate(MAX_ADVISORIES); + idx.security.retain_mut(|a| match a.validate() { + Err(why) => { + tracing::warn!(pkg = %a.pkg, "dropping advisory: {why}"); + false + } + Ok(()) => true, + }); + Ok(idx) + } +} + +impl Entry { + fn validate(&mut self) -> Result<()> { + if !valid_plugin_id(&self.id) { + bail!("id must be kebab-case `[a-z][a-z0-9-]*`, ≤64"); + } + if !valid_scoped_pkg(&self.pkg) { + bail!("pkg must be a scoped npm name (`@scope/name`)"); + } + if !is_https(&self.registry) { + bail!("registry must be an https:// URL"); + } + self.title = sanitize(&self.title, 64); + if self.title.is_empty() { + bail!("title must not be empty"); + } + self.description = sanitize(&self.description, 280); + self.author = sanitize(&self.author, 64); + self.version = sanitize(&self.version, 32); + if semver::Version::parse(&self.version).is_err() { + bail!("version must be exact semver (`1.2.3`), not a range"); + } + if !valid_integrity(&self.integrity) { + bail!("integrity must look like `sha512-`"); + } + if let Some(icon) = &self.icon { + if !valid_icon(icon) { + self.icon = None; // cosmetic — drop it rather than the whole entry + } + } + if let Some(h) = &self.homepage { + if !is_https(h) || h.len() > 200 { + self.homepage = None; + } + } + if let Some(l) = &self.license { + self.license = Some(sanitize(l, 64)).filter(|s| !s.is_empty()); + } + if let Some(v) = &self.verification { + if v.reviewed_at.len() > 32 { + self.verification = None; + } + } + if let Some(m) = &self.min_host { + if semver::Version::parse(m).is_err() { + self.min_host = None; // unusable constraint ⇒ no constraint + } + } + self.platforms + .retain(|p| matches!(p.as_str(), "linux" | "windows" | "macos")); + self.platforms.truncate(4); + Ok(()) + } + + /// Is this entry installable on the running host? Returns the operator-facing reason when not. + pub(crate) fn incompatible_reason(&self) -> Option { + if !self.platforms.is_empty() && !self.platforms.iter().any(|p| p == HOST_PLATFORM) { + return Some(format!("requires {}", self.platforms.join(" or "))); + } + if let Some(min) = &self.min_host { + let (Ok(min), Ok(host)) = ( + semver::Version::parse(min), + semver::Version::parse(host_version()), + ) else { + return None; + }; + if host < min { + return Some(format!("needs punktfunk {min} or newer")); + } + } + None + } +} + +impl Advisory { + fn validate(&mut self) -> Result<()> { + if self.pkg.trim().is_empty() || self.pkg.len() > 214 { + bail!("advisory pkg is empty or too long"); + } + semver::VersionReq::parse(&self.versions) + .context("advisory `versions` is not a semver requirement")?; + self.reason = sanitize(&self.reason, 280); + if let Some(u) = &self.url { + if !is_https(u) || u.len() > 200 { + self.url = None; + } + } + Ok(()) + } + + /// Does this advisory cover `pkg@version`? + pub(crate) fn matches(&self, pkg: &str, version: &str) -> bool { + if self.pkg != pkg { + return false; + } + let (Ok(req), Ok(v)) = ( + semver::VersionReq::parse(&self.versions), + semver::Version::parse(version), + ) else { + return false; + }; + req.matches(&v) + } +} + +// ---------------------------------------------------------------- small helpers + +/// This host's platform token, as used in [`Entry::platforms`]. +pub(crate) const HOST_PLATFORM: &str = if cfg!(target_os = "windows") { + "windows" +} else if cfg!(target_os = "macos") { + "macos" +} else { + "linux" +}; + +/// The running host's version — the left-hand side of every `minHost` comparison. +pub(crate) fn host_version() -> &'static str { + env!("CARGO_PKG_VERSION") +} + +/// Strip control characters (a catalog string ends up in a log line and in the console UI) and +/// clamp to `max` characters. +fn sanitize(s: &str, max: usize) -> String { + s.chars() + .filter(|c| !c.is_control()) + .take(max) + .collect::() + .trim() + .to_string() +} + +pub(crate) fn valid_plugin_id(id: &str) -> bool { + !id.is_empty() + && id.len() <= 64 + && id.as_bytes()[0].is_ascii_lowercase() + && id + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') +} + +/// `@scope/name` with npm's safe alphabet. Deliberately stricter than npm: no URL-ish characters +/// can survive into a `bun add` argument or a `bunfig.toml` scope key. +pub(crate) fn valid_scoped_pkg(pkg: &str) -> bool { + let Some(rest) = pkg.strip_prefix('@') else { + return false; + }; + if pkg.len() > 214 { + return false; + } + let Some((scope, name)) = rest.split_once('/') else { + return false; + }; + let ok = |s: &str| { + !s.is_empty() + && s.bytes().all(|b| { + b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'-' | b'_' | b'.') + }) + }; + ok(scope) && ok(name) +} + +/// The `@scope` half of a scoped package name — the key `bunfig.toml` maps to a registry. +pub(crate) fn scope_of(pkg: &str) -> Option { + let rest = pkg.strip_prefix('@')?; + let (scope, _) = rest.split_once('/')?; + Some(format!("@{scope}")) +} + +fn valid_icon(icon: &str) -> bool { + (1..=48).contains(&icon.len()) + && icon + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') +} + +/// npm's Subresource-Integrity spelling: `-`. +fn valid_integrity(s: &str) -> bool { + if s.len() > 200 { + return false; + } + let Some((alg, b64)) = s.split_once('-') else { + return false; + }; + matches!(alg, "sha512" | "sha384" | "sha256" | "sha1") + && !b64.is_empty() + && b64 + .bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'+' | b'/' | b'=')) +} + +fn is_https(url: &str) -> bool { + url.starts_with("https://") && url.len() > "https://".len() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn doc(entry_json: &str) -> Vec { + format!(r#"{{"schema":1,"name":"t","plugins":[{entry_json}]}}"#).into_bytes() + } + + const GOOD: &str = r#"{"id":"rom-manager","pkg":"@punktfunk/plugin-rom-manager", + "registry":"https://git.unom.io/api/packages/unom/npm/","title":"ROM Manager", + "description":"d","author":"unom","version":"0.2.1","integrity":"sha512-AAAA", + "verification":{"reviewedAt":"2026-07-19"},"platforms":["linux","windows"]}"#; + + #[test] + fn parses_a_good_entry() { + let idx = Index::parse(&doc(GOOD)).unwrap(); + assert_eq!(idx.plugins.len(), 1); + let e = &idx.plugins[0]; + assert_eq!(e.pkg, "@punktfunk/plugin-rom-manager"); + assert_eq!(e.version, "0.2.1"); + assert!(e.verification.is_some()); + } + + #[test] + fn rejects_unknown_schema_and_bad_json() { + assert!(Index::parse(br#"{"schema":2,"plugins":[]}"#).is_err()); + assert!(Index::parse(b"not json").is_err()); + } + + #[test] + fn drops_invalid_entries_but_keeps_the_rest() { + let bad_unscoped = GOOD.replace("@punktfunk/plugin-rom-manager", "punktfunk-plugin-x"); + let body = format!(r#"{{"schema":1,"plugins":[{bad_unscoped},{GOOD}]}}"#); + let idx = Index::parse(body.as_bytes()).unwrap(); + assert_eq!(idx.plugins.len(), 1, "unscoped pkg must be dropped (D8)"); + assert_eq!(idx.plugins[0].id, "rom-manager"); + } + + #[test] + fn rejects_version_ranges_and_http_registries() { + assert!(Index::parse(&doc( + &GOOD.replace(r#""version":"0.2.1""#, r#""version":"^0.2.1""#) + )) + .unwrap() + .plugins + .is_empty()); + assert!(Index::parse(&doc( + &GOOD.replace("https://git.unom.io", "http://git.unom.io") + )) + .unwrap() + .plugins + .is_empty()); + } + + #[test] + fn drops_duplicate_ids() { + let body = format!(r#"{{"schema":1,"plugins":[{GOOD},{GOOD}]}}"#); + assert_eq!(Index::parse(body.as_bytes()).unwrap().plugins.len(), 1); + } + + #[test] + fn sanitizes_control_characters_in_display_fields() { + let e = GOOD.replace("ROM Manager", "RO\\u0007M"); + let idx = Index::parse(&doc(&e)).unwrap(); + assert_eq!(idx.plugins[0].title, "ROM"); + } + + #[test] + fn advisory_matching_is_semver_ranged() { + let mut a = Advisory { + pkg: "@x/y".into(), + versions: "<0.3.2".into(), + reason: "bad".into(), + url: None, + }; + a.validate().unwrap(); + assert!(a.matches("@x/y", "0.3.1")); + assert!(!a.matches("@x/y", "0.3.2")); + assert!(!a.matches("@other/z", "0.3.1")); + } + + #[test] + fn advisory_with_unparseable_range_is_dropped() { + let body = r#"{"schema":1,"plugins":[],"security":[ + {"pkg":"@x/y","versions":"not a range","reason":"r"}]}"#; + assert!(Index::parse(body.as_bytes()).unwrap().security.is_empty()); + } + + #[test] + fn scope_extraction() { + assert_eq!(scope_of("@punktfunk/plugin-x").unwrap(), "@punktfunk"); + assert_eq!(scope_of("@a/b").unwrap(), "@a"); + assert!(scope_of("punktfunk-plugin-x").is_none()); + } + + #[test] + fn integrity_shape() { + assert!(valid_integrity("sha512-abcABC123+/=")); + assert!(!valid_integrity("md5-abc")); + assert!(!valid_integrity("sha512-")); + assert!(!valid_integrity("sha512")); + } + + #[test] + fn signature_verifies_only_over_exact_bytes_and_pinned_keys() { + use ring::signature::KeyPair as _; + let rng = ring::rand::SystemRandom::new(); + let pkcs8 = ring::signature::Ed25519KeyPair::generate_pkcs8(&rng).unwrap(); + let kp = ring::signature::Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap(); + let key = { + use base64::Engine as _; + PublicKey::parse(&format!( + "ed25519:{}", + base64::engine::general_purpose::STANDARD.encode(kp.public_key().as_ref()) + )) + .unwrap() + }; + let body = br#"{"schema":1,"plugins":[]}"#; + let sig = { + use base64::Engine as _; + base64::engine::general_purpose::STANDARD.encode(kp.sign(body).as_ref()) + }; + + assert!(verify_signature(body, &sig, std::slice::from_ref(&key)).is_ok()); + // whitespace in the .sig file is tolerated + assert!(verify_signature(body, &format!("{sig}\n"), std::slice::from_ref(&key)).is_ok()); + // one byte of tampering fails + assert!(verify_signature( + br#"{"schema":1,"plugins":[ ]}"#, + &sig, + std::slice::from_ref(&key) + ) + .is_err()); + // a different key fails + let other = ring::signature::Ed25519KeyPair::from_pkcs8( + ring::signature::Ed25519KeyPair::generate_pkcs8(&rng) + .unwrap() + .as_ref(), + ) + .unwrap(); + let other_key = { + use base64::Engine as _; + PublicKey::parse(&format!( + "ed25519:{}", + base64::engine::general_purpose::STANDARD.encode(other.public_key().as_ref()) + )) + .unwrap() + }; + assert!(verify_signature(body, &sig, &[other_key]).is_err()); + // no pinned key at all fails closed + assert!(verify_signature(body, &sig, &[]).is_err()); + } + + #[test] + fn public_key_parsing_rejects_junk() { + assert!(PublicKey::parse("nope").is_err()); + assert!(PublicKey::parse("ed25519:!!!").is_err()); + assert!(PublicKey::parse("ed25519:AAAA").is_err()); // wrong length + } +} diff --git a/crates/punktfunk-host/src/store/jobs.rs b/crates/punktfunk-host/src/store/jobs.rs new file mode 100644 index 00000000..89c737be --- /dev/null +++ b/crates/punktfunk-host/src/store/jobs.rs @@ -0,0 +1,687 @@ +//! Install / uninstall **jobs** (design `plugin-store.md` §4.3). +//! +//! A package operation takes tens of seconds, so the API hands back a job id immediately (202) and +//! the console polls. One at a time: `bun add` and `bun remove` share a lockfile and a +//! `node_modules` tree, so a second concurrent op is a corruption bug waiting to happen — a +//! request that arrives while a job runs gets 409, not a queue. +//! +//! The pipeline, and why each step is there: +//! +//! ```text +//! resolve → what exactly are we installing (catalog entry ⇒ pkg@version+integrity, or a raw spec) +//! verify → the registry's advertised integrity for that version MUST equal the index pin +//! install → spawn the runner CLI (`bun add`), streaming its output into the job log +//! check → the version on disk is the version we pinned, else roll back +//! record → provenance manifest, so the tier stays visible forever +//! restart → the runner rediscovers units only at startup, so activation is a restart +//! ``` +//! +//! **verify** is the step that makes "verified" mean something. A reviewed entry pins a tarball +//! hash; if the registry now advertises a different hash for that same version, someone republished +//! it after review and the install is refused before any code is fetched, let alone run. Tier-3 +//! (raw spec) installs skip it — there is no pin to check against, and that asymmetry *is* the +//! tier. + +use super::index::{scope_of, Entry}; +use super::manifest::{self, Record, Tier}; +use anyhow::{bail, Context, Result}; +use std::collections::VecDeque; +use std::io::{BufRead, BufReader, Read}; +use std::process::{Command, Stdio}; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +/// How long a package op may run before we kill it. `bun add` over a cold cache on a slow link is +/// the worst case; five minutes is far past it. +const JOB_TIMEOUT: Duration = Duration::from_secs(300); + +/// How many finished jobs we keep for the console to read back. +const JOB_HISTORY: usize = 20; + +/// Lines of runner output retained per job (the console shows a tail). +const LOG_LINES: usize = 200; + +// ---------------------------------------------------------------- job records + +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, utoipa::ToSchema)] +#[serde(rename_all = "lowercase")] +pub(crate) enum State { + Running, + Done, + Failed, +} + +/// A job as the console sees it. Field names are snake_case like the rest of the management API +/// (the *file* formats — index, sources, manifest — follow npm's camelCase instead). +#[derive(Debug, Clone, serde::Serialize, utoipa::ToSchema)] +pub(crate) struct Job { + pub id: String, + /// `install` or `uninstall`. + pub kind: String, + /// What the operator asked for — a package name, or the raw spec they typed. + pub target: String, + pub state: State, + /// Coarse step name, for a progress line the operator can read. + pub phase: String, + /// Tail of the runner's combined stdout/stderr. + pub log: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + pub started_at: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub finished_at: Option, +} + +struct Jobs { + jobs: VecDeque, + counter: u64, +} + +fn jobs() -> &'static Mutex { + static JOBS: std::sync::OnceLock> = std::sync::OnceLock::new(); + JOBS.get_or_init(|| { + Mutex::new(Jobs { + jobs: VecDeque::new(), + counter: 0, + }) + }) +} + +fn lock() -> std::sync::MutexGuard<'static, Jobs> { + jobs().lock().unwrap_or_else(|e| e.into_inner()) +} + +/// Start tracking a job, refusing if one is already in flight (single-flight, §4.3). +fn begin(kind: &str, target: &str) -> Result { + let mut g = lock(); + if let Some(active) = g.jobs.iter().find(|j| j.state == State::Running) { + bail!( + "another plugin operation is already running ({} {})", + active.kind, + active.target + ); + } + g.counter += 1; + let id = format!("job-{}-{}", super::catalog::unix_now(), g.counter); + let job = Job { + id: id.clone(), + kind: kind.to_string(), + target: target.to_string(), + state: State::Running, + phase: "queued".into(), + log: Vec::new(), + error: None, + started_at: super::catalog::unix_now(), + finished_at: None, + }; + g.jobs.push_back(job); + while g.jobs.len() > JOB_HISTORY { + g.jobs.pop_front(); + } + Ok(id) +} + +fn update(id: &str, f: impl FnOnce(&mut Job)) { + let mut g = lock(); + if let Some(job) = g.jobs.iter_mut().find(|j| j.id == id) { + f(job); + } +} + +fn set_phase(id: &str, phase: &str) { + tracing::info!(job = %id, phase, "plugin store job"); + update(id, |j| j.phase = phase.to_string()); +} + +fn log_line(id: &str, line: String) { + update(id, |j| { + if j.log.len() >= LOG_LINES { + j.log.remove(0); + } + j.log.push(line); + }); +} + +fn finish(id: &str, result: Result<()>) { + update(id, |j| { + j.finished_at = Some(super::catalog::unix_now()); + match &result { + Ok(()) => { + j.state = State::Done; + j.phase = "done".into(); + } + Err(e) => { + j.state = State::Failed; + j.error = Some(format!("{e:#}")); + } + } + }); + match result { + Ok(()) => tracing::info!(job = %id, "plugin store job finished"), + Err(e) => tracing::warn!(job = %id, "plugin store job failed: {e:#}"), + } +} + +/// One job by id. +pub(crate) fn get(id: &str) -> Option { + lock().jobs.iter().find(|j| j.id == id).cloned() +} + +/// Recent jobs, newest last. +pub(crate) fn list() -> Vec { + lock().jobs.iter().cloned().collect() +} + +/// Is a job in flight? (The console disables install buttons on this.) +pub(crate) fn busy() -> bool { + lock().jobs.iter().any(|j| j.state == State::Running) +} + +// ---------------------------------------------------------------- install plans + +/// A fully resolved install: everything the executor needs, with the trust decision already made. +pub(crate) struct Plan { + /// Package name without a version. + pub pkg: Option, + /// The argument handed to `bun add` — `pkg@version` for a catalogued entry, the raw spec + /// otherwise. + pub spec: String, + pub version: Option, + /// `(scope, registry_url)` to map in the plugins dir's `bunfig.toml`. + pub registry: Option<(String, String)>, + pub integrity: Option, + pub tier: Tier, + pub source: Option, + pub entry_id: Option, +} + +impl Plan { + /// From a catalog entry: exact version, pinned integrity, scope→registry mapping. + pub(crate) fn from_entry(entry: &Entry, source: &str, verified: bool) -> Result { + let scope = scope_of(&entry.pkg).context("catalog entry package must be scoped")?; + Ok(Plan { + pkg: Some(entry.pkg.clone()), + spec: format!("{}@{}", entry.pkg, entry.version), + version: Some(entry.version.clone()), + registry: Some((scope, entry.registry.clone())), + integrity: Some(entry.integrity.clone()), + tier: if verified { + Tier::Verified + } else { + Tier::External + }, + source: Some(source.to_string()), + entry_id: Some(entry.id.clone()), + }) + } + + /// From a raw package spec typed into the danger dialog. No pin, no review, no shelf. + pub(crate) fn from_spec(spec: &str) -> Result { + let spec = validate_spec(spec)?; + Ok(Plan { + pkg: parse_spec_pkg(&spec), + spec, + version: None, + registry: None, + integrity: None, + tier: Tier::Unverified, + source: None, + entry_id: None, + }) + } +} + +/// Accept only specs we're willing to hand to `bun add`. +/// +/// This is not shell quoting (we always exec with an argument vector, never a shell) — it is about +/// what the *package manager* will do with the string. Two real hazards: a leading `-` turns the +/// operand into a flag, and `file:` specs would let a console session install code from anywhere on +/// the host's filesystem, which is a different (and larger) capability than "install a package". +/// Local paths remain available through the CLI, which is where local development belongs. +fn validate_spec(spec: &str) -> Result { + let s = spec.trim(); + if s.is_empty() { + bail!("empty package spec"); + } + if s.len() > 400 { + bail!("package spec is too long"); + } + if s.starts_with('-') { + bail!("a package spec cannot start with '-'"); + } + if s.chars().any(|c| c.is_whitespace() || c.is_control()) { + bail!("a package spec cannot contain whitespace or control characters"); + } + let lower = s.to_ascii_lowercase(); + for bad in ["file:", "link:", "portal:"] { + if lower.starts_with(bad) { + bail!("`{bad}` specs are not installable from the console — use the `punktfunk-host plugins add` CLI"); + } + } + // URL-ish specs: only https and git+https. (http:// and git:// are unauthenticated transports.) + if lower.contains("://") + && !(lower.starts_with("https://") || lower.starts_with("git+https://")) + { + bail!("only https:// and git+https:// URLs are accepted"); + } + Ok(s.to_string()) +} + +/// Best-effort package name from an npm-style spec (`@scope/name`, `@scope/name@1.2.3`, `name@1`). +/// `None` for URL/git specs — those are identified after the fact by diffing `node_modules`. +fn parse_spec_pkg(spec: &str) -> Option { + if spec.contains("://") { + return None; + } + if let Some(rest) = spec.strip_prefix('@') { + // Scoped: the version separator is the '@' AFTER the scope's slash. + let (scope_and_name, _) = match rest.split_once('/') { + Some((scope, tail)) => match tail.split_once('@') { + Some((name, ver)) => (format!("@{scope}/{name}"), Some(ver)), + None => (format!("@{scope}/{tail}"), None), + }, + None => return None, + }; + return Some(scope_and_name); + } + Some( + spec.split_once('@') + .map(|(n, _)| n) + .unwrap_or(spec) + .to_string(), + ) +} + +// ---------------------------------------------------------------- execution + +/// Kick off an install. Returns the job id; the work runs on a detached thread. +pub(crate) fn spawn_install(plan: Plan) -> Result { + let target = plan.pkg.clone().unwrap_or_else(|| plan.spec.clone()); + let id = begin("install", &target)?; + let job_id = id.clone(); + std::thread::Builder::new() + .name("pf-store-install".into()) + .spawn(move || { + let result = run_install(&job_id, plan); + finish(&job_id, result); + crate::events::emit(crate::events::EventKind::StoreChanged); + }) + .context("spawn the plugin-install worker")?; + Ok(id) +} + +/// Kick off an uninstall. +pub(crate) fn spawn_uninstall(pkg: String) -> Result { + if super::valid_installed_pkg(&pkg).is_err() { + bail!("not an installable plugin package name"); + } + let id = begin("uninstall", &pkg)?; + let job_id = id.clone(); + std::thread::Builder::new() + .name("pf-store-uninstall".into()) + .spawn(move || { + let result = run_uninstall(&job_id, &pkg); + finish(&job_id, result); + crate::events::emit(crate::events::EventKind::StoreChanged); + }) + .context("spawn the plugin-uninstall worker")?; + Ok(id) +} + +fn run_install(id: &str, plan: Plan) -> Result<()> { + let dir = super::plugins_dir(); + + // ---- verify: the pin must still describe what the registry serves ------------------------ + if let (Some(integrity), Some(pkg), Some(version), Some((_, registry))) = ( + plan.integrity.as_deref(), + plan.pkg.as_deref(), + plan.version.as_deref(), + plan.registry.as_ref(), + ) { + set_phase(id, "verifying"); + let advertised = registry_integrity(registry, pkg, version) + .with_context(|| format!("check {pkg}@{version} against {registry}"))?; + if advertised != integrity { + bail!( + "integrity mismatch for {pkg}@{version}: the catalog pins {integrity} but the \ + registry now serves {advertised}. This version was republished after it was \ + reviewed — refusing to install." + ); + } + log_line(id, format!("integrity ok: {pkg}@{version}")); + } + + // ---- install ------------------------------------------------------------------------------ + set_phase(id, "installing"); + let before = super::installed_packages(&dir); + let mut args = vec!["add".to_string(), plan.spec.clone()]; + if plan.version.is_some() { + // Pin the dependency range too, so a later `bun install` in this tree can't drift off the + // reviewed version. + args.push("--exact".into()); + } + if let Some((scope, url)) = &plan.registry { + args.push("--registry".into()); + args.push(format!("{scope}={url}")); + } + if !plan.spec.starts_with("@punktfunk/") { + // The runner CLI's supply-chain gate refuses anything resolving off Punktfunk's own + // registry unless told otherwise. Here the operator already made that decision — either by + // adding the source, or in the danger dialog — so pass the flag rather than making the gate + // unable to express "yes, deliberately". + args.push("--allow-public-registry".into()); + } + args.push("--plugins".into()); + args.push(dir.to_string_lossy().into_owned()); + + run_runner(id, &args)?; + + // ---- check: is what landed what we asked for? --------------------------------------------- + set_phase(id, "checking"); + let after = super::installed_packages(&dir); + let added: Vec<_> = after + .iter() + .filter(|p| !before.iter().any(|b| b.pkg == p.pkg)) + .collect(); + // For a catalogued install we know the package name; for a URL/git spec we learn it here. + let pkg = plan + .pkg + .clone() + .or_else(|| added.first().map(|p| p.pkg.clone())) + .context( + "the install finished but no new plugin package appeared — is this package a \ + punktfunk plugin? (it must be named `@scope/plugin-*` or `punktfunk-plugin-*`)", + )?; + let installed = after + .iter() + .find(|p| p.pkg == pkg) + .with_context(|| format!("{pkg} is not present after install"))?; + + if let (Some(want), Some(got)) = (plan.version.as_deref(), installed.version.as_deref()) { + if want != got { + // Roll back rather than leave an unreviewed version installed under a verified badge. + log_line(id, format!("rolling back: expected {want}, found {got}")); + set_phase(id, "rolling back"); + let _ = run_runner( + id, + &[ + "remove".to_string(), + pkg.clone(), + "--plugins".to_string(), + dir.to_string_lossy().into_owned(), + ], + ); + bail!("installed {pkg}@{got} but the catalog pinned {want} — rolled back"); + } + } + + // ---- record + activate --------------------------------------------------------------------- + set_phase(id, "recording"); + manifest::record( + &dir, + &pkg, + Record { + tier: plan.tier, + source: plan.source.clone(), + entry_id: plan.entry_id.clone(), + version: installed.version.clone().or(plan.version.clone()), + spec: (plan.tier == Tier::Unverified).then(|| plan.spec.clone()), + installed_at: Some(manifest::now_stamp()), + }, + ) + .context("record install provenance")?; + + restart_runner(id); + Ok(()) +} + +fn run_uninstall(id: &str, pkg: &str) -> Result<()> { + let dir = super::plugins_dir(); + set_phase(id, "removing"); + run_runner( + id, + &[ + "remove".to_string(), + pkg.to_string(), + "--plugins".to_string(), + dir.to_string_lossy().into_owned(), + ], + )?; + set_phase(id, "recording"); + manifest::forget(&dir, pkg).context("update install provenance")?; + restart_runner(id); + Ok(()) +} + +/// Restart the runner so it rediscovers units. Best-effort and non-fatal: the package IS installed +/// at this point, so a restart failure must not present as an install failure — it presents as +/// "installed, runner needs a nudge", which the console says out loud. +fn restart_runner(id: &str) { + set_phase(id, "restarting runner"); + match crate::plugins::restart_runtime() { + Ok(true) => log_line(id, "plugin runner restarted".into()), + Ok(false) => log_line( + id, + "the plugin runner is not enabled — enable it to start this plugin".into(), + ), + Err(e) => log_line(id, format!("could not restart the plugin runner: {e:#}")), + } +} + +/// Run the bun runner CLI with `args`, streaming both streams into the job log. +fn run_runner(id: &str, args: &[String]) -> Result<()> { + let (program, prefix) = crate::plugins::runner_command()?; + tracing::info!(job = %id, program = %program.display(), ?args, "spawning the plugin runner"); + let mut child = Command::new(&program) + .args(&prefix) + .args(args) + // No stdin: `bun add` must never sit waiting on a prompt inside a service. + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .with_context(|| format!("run the plugin runner ({})", program.display()))?; + + // Pump both streams concurrently: bun writes progress to one and errors to the other, and a + // full pipe on either would block the child. + let mut pumps = Vec::new(); + if let Some(out) = child.stdout.take() { + let job = id.to_string(); + pumps.push(std::thread::spawn(move || pump(&job, out))); + } + if let Some(err) = child.stderr.take() { + let job = id.to_string(); + pumps.push(std::thread::spawn(move || pump(&job, err))); + } + + let deadline = Instant::now() + JOB_TIMEOUT; + let status = loop { + match child.try_wait().context("wait for the plugin runner")? { + Some(status) => break status, + None if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + bail!( + "the plugin runner timed out after {}s", + JOB_TIMEOUT.as_secs() + ); + } + None => std::thread::sleep(Duration::from_millis(150)), + } + }; + for p in pumps { + let _ = p.join(); + } + if !status.success() { + bail!( + "the plugin runner exited with status {} — see the job log", + status.code().unwrap_or(-1) + ); + } + Ok(()) +} + +/// Copy one child stream into the job log, line by line. +fn pump(job: &str, stream: impl std::io::Read) { + for line in BufReader::new(stream).lines().map_while(Result::ok) { + let line = line.trim_end().to_string(); + if !line.is_empty() { + log_line(job, line); + } + } +} + +// ---------------------------------------------------------------- registry preflight + +/// The integrity hash the registry itself advertises for `pkg@version`. +/// +/// This is the check that gives the pin teeth. npm clients already refuse a tarball whose bytes +/// don't match the registry's advertised hash, so the remaining attack is a *republish* — same +/// version, new content, new hash. Comparing the registry's current hash against the reviewed one +/// catches exactly that, before anything is downloaded. +fn registry_integrity(registry: &str, pkg: &str, version: &str) -> Result { + let base = registry.trim_end_matches('/'); + // npm registry convention: the scope separator is percent-encoded in the packument path. + let url = format!("{base}/{}", pkg.replace('/', "%2f")); + let agent = ureq::AgentBuilder::new() + .timeout(Duration::from_secs(20)) + .redirects(3) + .user_agent(&format!("punktfunk-host/{}", super::index::host_version())) + .build(); + let resp = agent + .get(&url) + .set("Accept", "application/json") + .call() + .map_err(|e| match e { + ureq::Error::Status(404, _) => { + anyhow::anyhow!("the registry does not know this package") + } + other => anyhow::anyhow!("registry request failed: {other}"), + })?; + let mut body = Vec::new(); + resp.into_reader() + .take(16 * 1024 * 1024) + .read_to_end(&mut body) + .context("read the registry response")?; + let doc: serde_json::Value = + serde_json::from_slice(&body).context("registry returned invalid JSON")?; + let dist = doc + .get("versions") + .and_then(|v| v.get(version)) + .and_then(|v| v.get("dist")) + .with_context(|| format!("the registry has no version {version} of this package"))?; + dist.get("integrity") + .and_then(|i| i.as_str()) + .map(str::to_string) + .context("the registry did not advertise an integrity hash for this version") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn spec_validation_refuses_the_dangerous_shapes() { + assert!(validate_spec("@scope/plugin-x").is_ok()); + assert!(validate_spec("@scope/plugin-x@1.2.3").is_ok()); + assert!(validate_spec("punktfunk-plugin-x").is_ok()); + assert!(validate_spec("https://e.org/p.tgz").is_ok()); + assert!(validate_spec("git+https://e.org/p.git").is_ok()); + + assert!(validate_spec("").is_err()); + assert!(validate_spec(" ").is_err()); + // a leading dash would be parsed by bun as a flag, not an operand + assert!(validate_spec("--production").is_err()); + assert!(validate_spec("a b").is_err()); + // local-path specs are CLI-only: installing from anywhere on the filesystem is a bigger + // capability than installing a package + assert!(validate_spec("file:/etc/passwd").is_err()); + assert!(validate_spec("link:../evil").is_err()); + // unauthenticated transports + assert!(validate_spec("http://e.org/p.tgz").is_err()); + assert!(validate_spec("git://e.org/p.git").is_err()); + } + + #[test] + fn spec_package_name_parsing() { + assert_eq!(parse_spec_pkg("@a/b").as_deref(), Some("@a/b")); + assert_eq!(parse_spec_pkg("@a/b@1.2.3").as_deref(), Some("@a/b")); + assert_eq!(parse_spec_pkg("plain").as_deref(), Some("plain")); + assert_eq!(parse_spec_pkg("plain@2").as_deref(), Some("plain")); + // URL specs are identified by diffing node_modules after the install, not by parsing + assert_eq!(parse_spec_pkg("https://e.org/p.tgz"), None); + } + + #[test] + fn plan_from_entry_pins_version_registry_and_integrity() { + let idx = super::super::index::Index::parse( + br#"{"schema":1,"plugins":[{"id":"rom-manager","pkg":"@punktfunk/plugin-rom-manager", + "registry":"https://git.unom.io/api/packages/unom/npm/","title":"ROM", + "version":"0.3.0","integrity":"sha512-AAAA"}]}"#, + ) + .unwrap(); + let plan = Plan::from_entry(&idx.plugins[0], "unom", true).unwrap(); + assert_eq!(plan.spec, "@punktfunk/plugin-rom-manager@0.3.0"); + assert_eq!(plan.version.as_deref(), Some("0.3.0")); + assert_eq!(plan.tier, Tier::Verified); + assert_eq!( + plan.registry.as_ref().unwrap().0, + "@punktfunk", + "the scope drives the bunfig registry mapping" + ); + assert_eq!(plan.integrity.as_deref(), Some("sha512-AAAA")); + + // The same entry from an operator-added source is External, never Verified (D6). + let ext = Plan::from_entry(&idx.plugins[0], "retro-hub", false).unwrap(); + assert_eq!(ext.tier, Tier::External); + assert_eq!(ext.source.as_deref(), Some("retro-hub")); + } + + #[test] + fn plan_from_spec_is_unverified_and_unpinned() { + let plan = Plan::from_spec(" @someone/punktfunk-plugin-x ").unwrap(); + assert_eq!(plan.tier, Tier::Unverified); + assert_eq!(plan.spec, "@someone/punktfunk-plugin-x"); + assert!(plan.version.is_none()); + assert!( + plan.integrity.is_none(), + "nothing to check a raw spec against" + ); + assert!(plan.registry.is_none()); + } + + /// `begin` is process-global and single-flight, so tests that create jobs must not overlap. + fn exclusive() -> std::sync::MutexGuard<'static, ()> { + static M: Mutex<()> = Mutex::new(()); + M.lock().unwrap_or_else(|e| e.into_inner()) + } + + #[test] + fn single_flight_refuses_a_second_job() { + let _g = exclusive(); + let first = begin("install", "@a/b").expect("first job starts"); + assert!(begin("install", "@c/d").is_err(), "second must be refused"); + assert!(busy()); + finish(&first, Ok(())); + assert!(!busy()); + let second = begin("install", "@c/d").expect("a finished job frees the slot"); + finish(&second, Ok(())); + } + + #[test] + fn job_log_is_bounded_and_tails() { + let _g = exclusive(); + let id = begin("install", "@log/test").unwrap(); + for i in 0..(LOG_LINES + 50) { + log_line(&id, format!("line {i}")); + } + let job = get(&id).unwrap(); + assert_eq!(job.log.len(), LOG_LINES); + assert_eq!(job.log.last().unwrap(), &format!("line {}", LOG_LINES + 49)); + finish(&id, Err(anyhow::anyhow!("boom"))); + let job = get(&id).unwrap(); + assert_eq!(job.state, State::Failed); + assert!(job.error.unwrap().contains("boom")); + assert!(job.finished_at.is_some()); + } +} diff --git a/crates/punktfunk-host/src/store/manifest.rs b/crates/punktfunk-host/src/store/manifest.rs new file mode 100644 index 00000000..9b7a7a59 --- /dev/null +++ b/crates/punktfunk-host/src/store/manifest.rs @@ -0,0 +1,246 @@ +//! The **provenance manifest**: how a plugin got onto this box, remembered for as long as it is +//! installed (design `plugin-store.md` §4.5). +//! +//! `node_modules` knows *what* is installed; it has no idea whether the operator picked a reviewed +//! entry off the official shelf or pasted a tarball URL into the danger dialog. That distinction is +//! exactly what the console has to keep showing — an unverified plugin must stay visibly +//! unverified in the Installed list and in the header above its own UI, long after the install +//! dialog is forgotten. +//! +//! `/install-manifest.json`, written by the host (which is the only thing that ever +//! performs a *store* install). **Absence is meaningful**: a package with no manifest entry was +//! installed by the CLI, and is reported as [`Tier::Cli`] rather than being silently promoted. + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; + +/// How a plugin came to be installed. Ordered by how much review stands behind it. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub(crate) enum Tier { + /// From the built-in source's index: unom reviewed this exact tarball. + Verified, + /// From an operator-added source's index: pinned and integrity-checked, but curated by + /// somebody else. Attribution, never the badge (D6). + External, + /// Installed from a raw package spec through the danger dialog. Nobody reviewed it. + Unverified, + /// No manifest entry — `punktfunk-host plugins add` put it there. + Cli, +} + +impl Tier { + pub(crate) fn as_str(self) -> &'static str { + match self { + Tier::Verified => "verified", + Tier::External => "external", + Tier::Unverified => "unverified", + Tier::Cli => "cli", + } + } +} + +/// One remembered install. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Record { + pub tier: Tier, + /// The catalog source's slug, when the install came off a shelf. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, + /// The catalog entry id, when there was one — lets the console re-associate an installed + /// package with its shelf row even if the package name is unusual. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub entry_id: Option, + /// The version we installed (as pinned). The *live* version always comes from disk; this is + /// what we asked for. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub version: Option, + /// The raw spec, for [`Tier::Unverified`] installs — the only record of what the operator + /// actually typed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub spec: Option, + /// RFC-3339 install time (wall clock — display only). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub installed_at: Option, +} + +#[derive(Serialize, Deserialize)] +struct ManifestFile { + schema: u32, + /// Keyed by npm package name. `BTreeMap` so the file has a stable, diffable order. + #[serde(default)] + plugins: BTreeMap, +} + +impl Default for ManifestFile { + fn default() -> Self { + Self { + schema: 1, + plugins: BTreeMap::new(), + } + } +} + +fn manifest_path(plugins_dir: &std::path::Path) -> std::path::PathBuf { + plugins_dir.join("install-manifest.json") +} + +/// Read the manifest. A missing or corrupt file reads as empty — every package then reports as +/// CLI-installed, which is the honest answer when we have no provenance to show. +pub(crate) fn load(plugins_dir: &std::path::Path) -> BTreeMap { + let Ok(bytes) = std::fs::read(manifest_path(plugins_dir)) else { + return BTreeMap::new(); + }; + match serde_json::from_slice::(&bytes) { + Ok(f) if f.schema == 1 => f.plugins, + Ok(f) => { + tracing::warn!( + schema = f.schema, + "unknown install-manifest schema — ignoring" + ); + BTreeMap::new() + } + Err(e) => { + tracing::warn!("install-manifest.json is unreadable ({e}) — treating as empty"); + BTreeMap::new() + } + } +} + +/// Record (or replace) one package's provenance. +pub(crate) fn record(plugins_dir: &std::path::Path, pkg: &str, rec: Record) -> Result<()> { + let mut plugins = load(plugins_dir); + plugins.insert(pkg.to_string(), rec); + write(plugins_dir, plugins) +} + +/// Forget a package (on uninstall). Silent if it wasn't recorded. +pub(crate) fn forget(plugins_dir: &std::path::Path, pkg: &str) -> Result<()> { + let mut plugins = load(plugins_dir); + if plugins.remove(pkg).is_none() { + return Ok(()); + } + write(plugins_dir, plugins) +} + +fn write(plugins_dir: &std::path::Path, plugins: BTreeMap) -> Result<()> { + std::fs::create_dir_all(plugins_dir) + .with_context(|| format!("create {}", plugins_dir.display()))?; + let json = serde_json::to_string_pretty(&ManifestFile { schema: 1, plugins }) + .context("serialize install-manifest.json")?; + let path = manifest_path(plugins_dir); + let tmp = path.with_extension("json.tmp"); + std::fs::write(&tmp, format!("{json}\n")) + .with_context(|| format!("write {}", tmp.display()))?; + std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?; + Ok(()) +} + +/// Wall-clock RFC-3339-ish stamp for [`Record::installed_at`]. Display only — nothing compares +/// these, so a clock jump costs a cosmetic oddity, never a decision. +pub(crate) fn now_stamp() -> String { + let secs = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + // Civil-date arithmetic from a Unix timestamp (days since epoch → y/m/d), so this needs no + // date crate. UTC, second resolution. + let (days, rem) = ((secs / 86_400) as i64, secs % 86_400); + let (h, mi, s) = (rem / 3600, (rem % 3600) / 60, rem % 60); + let z = days + 719_468; + let era = z.div_euclid(146_097); + let doe = z.rem_euclid(146_097); + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let y = yoe + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let m = if mp < 10 { mp + 3 } else { mp - 9 }; + let y = if m <= 2 { y + 1 } else { y }; + format!("{y:04}-{m:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z") +} + +#[cfg(test)] +mod tests { + use super::*; + + fn rec(tier: Tier) -> Record { + Record { + tier, + source: Some("unom".into()), + entry_id: Some("rom-manager".into()), + version: Some("0.2.1".into()), + spec: None, + installed_at: Some(now_stamp()), + } + } + + #[test] + fn round_trips_and_absence_means_cli() { + let dir = tempfile::tempdir().unwrap(); + assert!(load(dir.path()).is_empty(), "no file ⇒ nothing recorded"); + + record( + dir.path(), + "@punktfunk/plugin-rom-manager", + rec(Tier::Verified), + ) + .unwrap(); + let m = load(dir.path()); + let r = m.get("@punktfunk/plugin-rom-manager").unwrap(); + assert_eq!(r.tier, Tier::Verified); + assert_eq!(r.version.as_deref(), Some("0.2.1")); + // A package we never recorded has no entry — the caller reports it as CLI-installed. + assert!(m.get("punktfunk-plugin-other").is_none()); + } + + #[test] + fn tier_survives_a_reinstall_at_a_new_tier() { + let dir = tempfile::tempdir().unwrap(); + record(dir.path(), "@x/y", rec(Tier::Verified)).unwrap(); + record(dir.path(), "@x/y", rec(Tier::Unverified)).unwrap(); + assert_eq!(load(dir.path()).get("@x/y").unwrap().tier, Tier::Unverified); + } + + #[test] + fn forget_removes_only_the_named_package() { + let dir = tempfile::tempdir().unwrap(); + record(dir.path(), "@x/y", rec(Tier::Verified)).unwrap(); + record(dir.path(), "@x/z", rec(Tier::External)).unwrap(); + forget(dir.path(), "@x/y").unwrap(); + let m = load(dir.path()); + assert!(m.get("@x/y").is_none()); + assert!(m.get("@x/z").is_some()); + forget(dir.path(), "@not/here").unwrap(); // no-op, no error + } + + #[test] + fn corrupt_manifest_reads_as_empty_not_an_error() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("install-manifest.json"), b"{ not json").unwrap(); + assert!(load(dir.path()).is_empty()); + } + + #[test] + fn tier_json_spelling_is_stable() { + // The console keys its badges off these strings. + assert_eq!( + serde_json::to_string(&Tier::Unverified).unwrap(), + "\"unverified\"" + ); + assert_eq!(Tier::Verified.as_str(), "verified"); + } + + #[test] + fn stamp_is_rfc3339_shaped() { + let s = now_stamp(); + assert_eq!(s.len(), 20, "{s}"); + assert!(s.ends_with('Z')); + assert_eq!(&s[4..5], "-"); + assert_eq!(&s[10..11], "T"); + // A known epoch value, to catch the civil-date arithmetic drifting. + assert!(s > "2026-01-01T00:00:00Z".to_string(), "{s}"); + } +} diff --git a/crates/punktfunk-host/src/store/sources.rs b/crates/punktfunk-host/src/store/sources.rs new file mode 100644 index 00000000..216d1163 --- /dev/null +++ b/crates/punktfunk-host/src/store/sources.rs @@ -0,0 +1,286 @@ +//! Catalog **sources** — where the store's shelves come from (design `plugin-store.md` §3.1). +//! +//! One source = one URL serving a signed index. The **official** source is compiled in (URL plus +//! two pinned key slots) and cannot be edited or removed; operator-added sources live in +//! `/plugin-sources.json`. +//! +//! Adding a source is itself a trust decision, made once: a source's entries become installable +//! on this host. That is why they carry attribution and warning styling in the console but never +//! the "Verified" badge — verification is unom reviewing a specific tarball, and it is not +//! transferable to a third-party curator (D6). + +use super::index::PublicKey; +use anyhow::{bail, Context, Result}; +use serde::{Deserialize, Serialize}; + +/// The built-in source's slug. Reserved: an operator source may not take this name. +pub(crate) const OFFICIAL_NAME: &str = "unom"; + +/// The built-in source's index URL. +pub(crate) const OFFICIAL_URL: &str = "https://plugins.punktfunk.unom.io/v1/index.json"; + +/// Pinned signing keys for the built-in source. **Two slots** so a key rotation is "sign with the +/// new key, ship a host that trusts both, retire the old one" instead of a flag day where old +/// hosts lose the catalog. An empty slot is ignored. +pub(crate) const OFFICIAL_KEYS: [&str; 2] = [ + "ed25519:n/KgBQSSDZqvyGHa8PkHWHZtV1zRXLk0BdQUi4/BD/w=", + "", // rotation slot +]; + +/// How many operator sources we'll hold. A guard rail, not a design limit. +const MAX_SOURCES: usize = 32; + +/// A catalog source as stored in `plugin-sources.json`. camelCase like the index document it +/// points at (the management API's own view of a source is snake_case — see `mgmt::store`). +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Source { + /// Slug (`[a-z][a-z0-9-]*`, ≤32) — the cache-file name and the console's attribution chip. + pub name: String, + /// `https://` URL of the index document. The `.sig` is fetched from `.sig`. + pub url: String, + /// Pinned ed25519 key. A source without one is accepted but marked **unsigned**, and its + /// entries inherit that marker — we can still show you the shelf, we just can't promise the + /// shelf wasn't rewritten in transit. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub public_key: Option, +} + +impl Source { + /// The compiled-in official source. + pub(crate) fn official() -> Source { + Source { + name: OFFICIAL_NAME.to_string(), + url: OFFICIAL_URL.to_string(), + // Held in [`OFFICIAL_KEYS`], not here: the built-in keys are code, not config, so a + // config file can never downgrade the official source to unsigned. + public_key: None, + } + } + + pub(crate) fn is_official(&self) -> bool { + self.name == OFFICIAL_NAME + } + + /// The keys this source's index must verify against. Empty ⇒ unsigned source (accepted, but + /// flagged); for the official source this is always the compiled-in pair. + pub(crate) fn keys(&self) -> Vec { + if self.is_official() { + return OFFICIAL_KEYS + .iter() + .filter(|k| !k.is_empty()) + .filter_map(|k| PublicKey::parse(k).ok()) + .collect(); + } + self.public_key + .as_deref() + .and_then(|k| PublicKey::parse(k).ok()) + .into_iter() + .collect() + } + + /// Does this source's index carry a signature we check? Drives the console's "unsigned" marker. + pub(crate) fn is_signed(&self) -> bool { + !self.keys().is_empty() + } + + /// Where the detached signature lives. Kept a pure function of the index URL so a source + /// record can never point the signature fetch somewhere else than the document it signs. + pub(crate) fn sig_url(&self) -> String { + format!("{}.sig", self.url) + } + + fn validate(&self) -> Result<()> { + if !valid_source_name(&self.name) { + bail!("source name must be kebab-case `[a-z][a-z0-9-]*`, ≤32 characters"); + } + if !self.url.starts_with("https://") || self.url.len() > 500 { + bail!("source url must be an https:// URL (≤500 characters)"); + } + if let Some(k) = &self.public_key { + PublicKey::parse(k).context("source publicKey")?; + } + Ok(()) + } +} + +pub(crate) fn valid_source_name(name: &str) -> bool { + !name.is_empty() + && name.len() <= 32 + && name.as_bytes()[0].is_ascii_lowercase() + && name + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') +} + +// ---------------------------------------------------------------- persistence + +#[derive(Serialize, Deserialize, Default)] +struct SourcesFile { + #[serde(default = "one")] + schema: u32, + #[serde(default)] + sources: Vec, +} + +fn one() -> u32 { + 1 +} + +fn sources_path() -> std::path::PathBuf { + pf_paths::config_dir().join("plugin-sources.json") +} + +/// Every source this host reads, official first. The official source is prepended here rather +/// than stored, so it survives a hand-edited (or deleted) config file. +pub(crate) fn load() -> Vec { + let mut out = vec![Source::official()]; + let path = sources_path(); + let Ok(bytes) = std::fs::read(&path) else { + return out; // no operator sources yet — the common case + }; + match serde_json::from_slice::(&bytes) { + Ok(file) => { + for s in file.sources.into_iter().take(MAX_SOURCES) { + // Defense in depth: a hand-edited file can't smuggle in a source that shadows the + // official one or carries an unusable URL/key. + if s.is_official() || s.validate().is_err() { + tracing::warn!(name = %s.name, "ignoring invalid entry in plugin-sources.json"); + continue; + } + if !out.iter().any(|e| e.name == s.name) { + out.push(s); + } + } + } + Err(e) => tracing::warn!( + "plugin-sources.json is unreadable ({e}) — using the official source only" + ), + } + out +} + +/// Add or replace an operator source. Refuses the reserved official name and invalid records. +pub(crate) fn put(source: Source) -> Result<()> { + source.validate()?; + if source.is_official() { + bail!("`{OFFICIAL_NAME}` is the built-in source and cannot be redefined"); + } + let mut list: Vec = load().into_iter().filter(|s| !s.is_official()).collect(); + if list.len() >= MAX_SOURCES && !list.iter().any(|s| s.name == source.name) { + bail!("too many plugin sources (max {MAX_SOURCES})"); + } + list.retain(|s| s.name != source.name); + list.push(source); + save(list) +} + +/// Remove an operator source. Returns `false` if there was nothing to remove. +pub(crate) fn remove(name: &str) -> Result { + if name == OFFICIAL_NAME { + bail!("the built-in `{OFFICIAL_NAME}` source cannot be removed"); + } + let list: Vec = load().into_iter().filter(|s| !s.is_official()).collect(); + if !list.iter().any(|s| s.name == name) { + return Ok(false); // nothing to remove — don't rewrite the file + } + save(list.into_iter().filter(|s| s.name != name).collect())?; + Ok(true) +} + +fn save(list: Vec) -> Result<()> { + let dir = pf_paths::config_dir(); + pf_paths::create_private_dir(&dir).context("create the punktfunk config dir")?; + let file = SourcesFile { + schema: 1, + sources: list, + }; + let json = serde_json::to_string_pretty(&file).context("serialize plugin-sources.json")?; + let path = sources_path(); + // Write-then-rename: a crash mid-write must not leave a half-file that makes the host forget + // the operator's sources on next boot. + let tmp = path.with_extension("json.tmp"); + std::fs::write(&tmp, format!("{json}\n")) + .with_context(|| format!("write {}", tmp.display()))?; + std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn official_source_is_signed_by_compiled_in_keys() { + let s = Source::official(); + assert!(s.is_official()); + assert!(s.is_signed(), "the built-in source must carry a pinned key"); + assert_eq!(s.sig_url(), format!("{OFFICIAL_URL}.sig")); + // A config file cannot downgrade it: `keys()` ignores the record's own field entirely. + let mut forged = Source::official(); + forged.public_key = Some("ed25519:AAAA".into()); + assert_eq!(forged.keys().len(), 1); + } + + #[test] + fn compiled_in_keys_parse() { + for k in OFFICIAL_KEYS.iter().filter(|k| !k.is_empty()) { + PublicKey::parse(k).expect("compiled-in official key must parse"); + } + } + + #[test] + fn unsigned_third_party_source_is_flagged_not_rejected() { + let s = Source { + name: "retro-hub".into(), + url: "https://example.org/index.json".into(), + public_key: None, + }; + s.validate().unwrap(); + assert!(!s.is_signed()); + } + + #[test] + fn validation_rejects_bad_names_urls_and_keys() { + let base = Source { + name: "ok".into(), + url: "https://e.org/i.json".into(), + public_key: None, + }; + assert!(base.validate().is_ok()); + assert!(Source { + name: "Bad".into(), + ..base.clone() + } + .validate() + .is_err()); + assert!(Source { + name: "9bad".into(), + ..base.clone() + } + .validate() + .is_err()); + assert!(Source { + url: "http://e.org/i.json".into(), + ..base.clone() + } + .validate() + .is_err()); + assert!(Source { + public_key: Some("ed25519:nope".into()), + ..base.clone() + } + .validate() + .is_err()); + } + + #[test] + fn sig_url_is_derived_not_configurable() { + let s = Source { + name: "x".into(), + url: "https://e.org/v1/index.json".into(), + public_key: None, + }; + assert_eq!(s.sig_url(), "https://e.org/v1/index.json.sig"); + } +} diff --git a/sdk/src/plugins.ts b/sdk/src/plugins.ts index fe066d4f..50628588 100644 --- a/sdk/src/plugins.ts +++ b/sdk/src/plugins.ts @@ -61,42 +61,75 @@ export const ensurePluginsDir = (dir = pluginsDirDefault()): string => { }; /** - * Ensure `/bunfig.toml` points the `@punktfunk` scope at the registry so `bun add` resolves - * first-party plugins. Idempotent: a file already mapping the scope is left untouched; an existing - * bunfig with an `[install.scopes]` table gets our line inserted under it; anything else appends a - * fresh table. + * Ensure `/bunfig.toml` maps every scope we need to its registry, so `bun add` resolves + * plugins from the right place. `@punktfunk` → Punktfunk's own registry is always mapped; + * `extraScopes` adds others — a plugin-store catalog entry carries its own registry, and the scope + * is what binds a package name to it (design D8, which is why catalog entries must be scoped). + * + * Idempotent and non-destructive: a scope already mapped to the same URL is left alone, a scope + * mapped to a *different* URL is rewritten, and any unrelated bunfig content is preserved. */ -export const ensureBunfig = (dir = pluginsDirDefault()): void => { +export const ensureBunfig = ( + dir = pluginsDirDefault(), + extraScopes: Record = {}, +): void => { const file = path.join(dir, "bunfig.toml"); - const scopeLine = `"@punktfunk" = "${REGISTRY}"`; + const wanted: Record = { "@punktfunk": REGISTRY, ...extraScopes }; let existing = ""; try { existing = fs.readFileSync(file, "utf8"); } catch { // no bunfig yet — write a fresh one below } - if (existing.includes("@punktfunk") && existing.includes(REGISTRY)) return; // already wired - const table = `[install.scopes]\n${scopeLine}\n`; - if (!existing.trim()) { - fs.writeFileSync(file, table); - } else if (/^\[install\.scopes\][^\n]*$/m.test(existing)) { - // Insert our scope line right after the existing table header. + let out = existing; + const missing: string[] = []; + for (const [scope, url] of Object.entries(wanted)) { + // Match `"@scope" = "…"` (quoted or bare key) anywhere in the file. + const line = new RegExp(`^\\s*"?${escapeRe(scope)}"?\\s*=\\s*".*"\\s*$`, "m"); + const replacement = `"${scope}" = "${url}"`; + if (line.test(out)) { + const current = out.match(line)?.[0] ?? ""; + if (current.includes(`"${url}"`)) continue; // already correct + out = out.replace(line, replacement); + } else { + missing.push(replacement); + } + } + if (missing.length === 0) { + if (out !== existing) fs.writeFileSync(file, out); + return; + } + const block = missing.join("\n"); + if (!out.trim()) { + fs.writeFileSync(file, `[install.scopes]\n${block}\n`); + } else if (/^\[install\.scopes\][^\n]*$/m.test(out)) { + // Insert under the existing table header. fs.writeFileSync( file, - existing.replace(/^\[install\.scopes\][^\n]*$/m, (m) => `${m}\n${scopeLine}`), + out.replace(/^\[install\.scopes\][^\n]*$/m, (m) => `${m}\n${block}`), ); } else { - const sep = existing.endsWith("\n") ? "" : "\n"; - fs.writeFileSync(file, `${existing}${sep}\n${table}`); + const sep = out.endsWith("\n") ? "" : "\n"; + fs.writeFileSync(file, `${out}${sep}\n[install.scopes]\n${block}\n`); } }; +const escapeRe = (s: string): string => s.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); + export interface PkgOpts extends ResolveOptions { /** Plugins dir. Default `/plugins`. */ dir?: string; /** Line sink for progress. Default stdout. */ log?: (line: string) => void; + /** + * Record the resolved version exactly (`bun add --exact`) instead of a caret range. The plugin + * store always sets this: a catalog entry pins one reviewed version, and a caret range in + * `package.json` would let a later `bun install` in this tree drift off it. + */ + exact?: boolean; + /** Extra `scope → registry URL` mappings to write into `bunfig.toml` before installing. */ + registries?: Record; } /** Run `bun add`/`bun remove` in the plugins dir on the current (vendored) bun. */ @@ -104,11 +137,21 @@ const runBun = (action: "add" | "remove", pkgs: string[], opts: PkgOpts): void = const dir = opts.dir ?? pluginsDirDefault(); const log = opts.log ?? ((l: string) => console.log(l)); ensurePluginsDir(dir); - if (action === "add") ensureBunfig(dir); + if (action === "add") ensureBunfig(dir, opts.registries); log(`${action === "add" ? "installing" : "removing"} ${pkgs.join(", ")} in ${dir}`); // `process.execPath` is the bun running this file (the vendored one under the package), so a // system-wide bun on PATH is not required. Inherit stdio so `bun`'s progress reaches the user. const args = [process.execPath, action, ...pkgs]; + if (action === "add") { + // NEVER run install lifecycle scripts. A plugin is code we chose to run under the runner, + // where it is supervised and (on Windows) de-privileged; a postinstall script runs + // immediately, as whoever is installing — which on a console-triggered install is the host + // service. bun already declines untrusted scripts by default; this makes it explicit and + // unconditional. A plugin that needs a native build step is a review rejection, not a case + // to support. + args.push("--ignore-scripts"); + if (opts.exact) args.push("--exact"); + } // Windows: install file COPIES, never bun's default hardlinks. A hardlinked file's canonical // path resolves into the installing admin's per-user bun cache // (C:\Users\\.bun\install\cache\…), which the de-privileged LocalService runner cannot @@ -156,9 +199,10 @@ export interface InstalledPlugin { } /** - * Enumerate installed plugin packages under `/node_modules` — both the scoped first-party - * convention (`@punktfunk/plugin-*`) and the unscoped one (`punktfunk-plugin-*`). Mirrors the - * discovery in runner.ts so `list` shows exactly what the runner would supervise. + * Enumerate installed plugin packages under `/node_modules` — the unscoped convention + * (`punktfunk-plugin-*`) and **any** scope's `plugin-*` (`@punktfunk/plugin-rom-manager`, + * `@retro-hub/plugin-x`). Mirrors the discovery in runner.ts so `list` shows exactly what the + * runner would supervise. */ export const listInstalled = (dir = pluginsDirDefault()): InstalledPlugin[] => { const modules = path.join(dir, "node_modules"); @@ -182,7 +226,7 @@ export const listInstalled = (dir = pluginsDirDefault()): InstalledPlugin[] => { for (const entry of entries) { if (entry.startsWith("punktfunk-plugin-")) { out.push({ pkg: entry, version: versionOf(path.join(modules, entry)) }); - } else if (entry === "@punktfunk") { + } else if (entry.startsWith("@")) { let scoped: string[] = []; try { scoped = fs.readdirSync(path.join(modules, entry)).sort(); diff --git a/sdk/src/runner-cli.ts b/sdk/src/runner-cli.ts index d8762087..00ac2cff 100644 --- a/sdk/src/runner-cli.ts +++ b/sdk/src/runner-cli.ts @@ -16,6 +16,11 @@ // // bun src/runner-cli.ts [--scripts DIR] [--plugins DIR] [--list] (run the runner) // bun src/runner-cli.ts add playnite [--plugins DIR] (package ops) +// +// Package-op flags: --exact pins the resolved version instead of a caret range, and +// --registry @scope=https://… maps a scope to its registry in bunfig.toml. Both exist for the +// plugin store (crates/punktfunk-host/src/store), which installs one reviewed version of a +// package that may live on somebody else's registry — but they are ordinary CLI flags too. import { Effect, Fiber } from "effect"; import { addPlugins, listInstalled, removePlugins } from "./plugins.js"; import { discoverUnits, runner } from "./runner.js"; @@ -25,18 +30,56 @@ const arg = (flag: string): string | undefined => { return i >= 0 ? process.argv[i + 1] : undefined; }; +/** Every value of a repeatable flag (`--registry a=b --registry c=d`). */ +const argAll = (flag: string): string[] => { + const out: string[] = []; + for (let i = 0; i < process.argv.length; i++) { + if (process.argv[i] === flag && process.argv[i + 1] !== undefined) out.push(process.argv[++i]); + } + return out; +}; + +/** Flags that take a value — skipped (with their value) when collecting positionals. */ +const VALUE_FLAGS = ["--plugins", "--scripts", "--registry"]; + const options = { scriptsDir: arg("--scripts"), pluginsDir: arg("--plugins"), }; +/** + * `--registry @scope=https://registry/` → `{ "@scope": "https://registry/" }`. The plugin store + * passes one per catalog entry so a third-party package's scope resolves to *its* registry rather + * than the public npm default. + */ +const parseRegistries = (): Record => { + const out: Record = {}; + for (const spec of argAll("--registry")) { + const eq = spec.indexOf("="); + if (eq <= 0) { + console.error(`[plugins] ignoring malformed --registry '${spec}' (expected @scope=URL)`); + continue; + } + const scope = spec.slice(0, eq).trim(); + const url = spec.slice(eq + 1).trim(); + if (!scope.startsWith("@") || !url.startsWith("https://")) { + console.error( + `[plugins] ignoring --registry '${spec}' (scope must start with @, url with https://)`, + ); + continue; + } + out[scope] = url; + } + return out; +}; + // Positional plugin names after the subcommand (argv: [bun, script, , …]). Skip flags and the // value of `--plugins`/`--scripts` wherever they appear, so ordering doesn't matter. const positionals = (): string[] => { const out: string[] = []; for (let i = 3; i < process.argv.length; i++) { const a = process.argv[i]; - if (a === "--plugins" || a === "--scripts") { + if (VALUE_FLAGS.includes(a)) { i++; // skip its value too continue; } @@ -51,6 +94,10 @@ const pkgOpts = { // Opt-in for names that resolve on the public npm registry (supply-chain gate in // plugins.ts::resolvePackage). Boolean flag, so positionals() skips it on its own. allowPublicRegistry: process.argv.includes("--allow-public-registry"), + // Pin the exact version in package.json instead of a caret range — the plugin store installs + // one reviewed version and must not let a later `bun install` drift off it. + exact: process.argv.includes("--exact"), + registries: parseRegistries(), }; const runPkgOp = ( diff --git a/sdk/src/runner.ts b/sdk/src/runner.ts index 02317038..b2ab539d 100644 --- a/sdk/src/runner.ts +++ b/sdk/src/runner.ts @@ -323,10 +323,15 @@ export const discoverUnits = ( addPlugin(path.join(modules, pkg), pkg); continue; } - // Scoped convention: `@punktfunk/plugin-*` (first-party). A scoped name resolves cleanly - // from a single registry scope-map, so a plugin can depend on `@punktfunk/host` + `effect` - // as shared (hoisted) deps rather than bundling its own copy of each. - if (pkg === "@punktfunk") { + // Scoped convention: `/plugin-*`. A scoped name resolves cleanly from a + // registry scope-map, so a plugin can depend on `@punktfunk/host` + `effect` as shared + // (hoisted) deps rather than bundling its own copy of each. + // + // ANY scope, not just `@punktfunk`: the plugin store requires catalog entries to be + // scoped precisely so the scope can map to that entry's registry, so a third-party + // plugin necessarily arrives as `@their-scope/plugin-*`. Limiting discovery to the + // first-party scope would let such a plugin install and then never run. + if (pkg.startsWith("@")) { try { for (const scoped of fs.readdirSync(path.join(modules, pkg)).sort()) { if (scoped.startsWith("plugin-")) { @@ -334,7 +339,7 @@ export const discoverUnits = ( } } } catch { - // no @punktfunk scope dir — fine + // not a readable scope dir — fine } } } diff --git a/sdk/test/plugins.test.ts b/sdk/test/plugins.test.ts index adc2554a..a2fae353 100644 --- a/sdk/test/plugins.test.ts +++ b/sdk/test/plugins.test.ts @@ -136,6 +136,55 @@ describe("listInstalled", () => { { pkg: "punktfunk-plugin-broken", version: undefined }, ]); }); + + test("finds plugins in ANY scope, not just @punktfunk", () => { + // Plugin-store catalog entries must be scoped so the scope can map to that entry's + // registry, so a third-party plugin necessarily arrives as `@their-scope/plugin-*`. + // Discovery limited to @punktfunk would let it install and then never run. + const dir = tmp("list-foreign-scope"); + writePkg(dir, path.join("@retro-hub", "plugin-x"), "1.0.0"); + writePkg(dir, path.join("@retro-hub", "helper"), "1.0.0"); // scoped non-plugin — ignored + expect(listInstalled(dir)).toEqual([{ pkg: "@retro-hub/plugin-x", version: "1.0.0" }]); + }); +}); + +describe("ensureBunfig with extra scopes", () => { + const read = (dir: string) => fs.readFileSync(path.join(dir, "bunfig.toml"), "utf8"); + + test("maps a third-party scope alongside @punktfunk", () => { + const dir = tmp("bunfig-extra"); + ensureBunfig(dir, { "@retro-hub": "https://retro.example/npm/" }); + const out = read(dir); + expect(out).toContain(`"@punktfunk" = "${REGISTRY}"`); + expect(out).toContain('"@retro-hub" = "https://retro.example/npm/"'); + }); + + test("is idempotent and rewrites a scope whose registry changed", () => { + const dir = tmp("bunfig-rewrite"); + ensureBunfig(dir, { "@retro-hub": "https://old.example/npm/" }); + ensureBunfig(dir, { "@retro-hub": "https://old.example/npm/" }); + expect(read(dir).match(/@retro-hub/g)?.length).toBe(1); + + ensureBunfig(dir, { "@retro-hub": "https://new.example/npm/" }); + const out = read(dir); + expect(out).toContain('"@retro-hub" = "https://new.example/npm/"'); + expect(out).not.toContain("old.example"); + // the first-party mapping survives an unrelated scope edit + expect(out).toContain(`"@punktfunk" = "${REGISTRY}"`); + }); + + test("preserves unrelated scopes already in the file", () => { + const dir = tmp("bunfig-preserve"); + fs.writeFileSync( + path.join(dir, "bunfig.toml"), + '[install.scopes]\n"@acme" = "https://acme.example/"\n', + ); + ensureBunfig(dir, { "@retro-hub": "https://retro.example/npm/" }); + const out = read(dir); + expect(out).toContain('"@acme" = "https://acme.example/"'); + expect(out).toContain('"@retro-hub" = "https://retro.example/npm/"'); + expect(out).toContain(`"@punktfunk" = "${REGISTRY}"`); + }); }); describe("ensurePluginsDir", () => {