feat: real plugin system with loadable instances + OpenBao secrets (v1.17.0)
Generalize the half-built discovery plugins into a real plugin system: plugin TYPES (the plugins/<category>/<type>.js modules with manifests) and loadable, configurable, multi-copy plugin INSTANCES (PluginInstance ORM model) managed from a dedicated /plugins page and /api/plugins API, with per-instance secrets in OpenBao at secret/plugins/<id>/conf. - plugin_registry.js: getTypes/getModule/splitConfig/mask + required-field helpers - PluginInstance model (Sequelize): id/pluginType/category/name/slug(unique)/ enabled/cron/config(json, non-secret)/lastRun*; registered in models/index.js - plugin_secrets.js: read/write/remove/mergeForRun over @simpleworkjs/bao-conf - scheduler.js: schedules from the DB registry; per-instance stable BullMQ JobScheduler ids (plugin:<id>) for load/unload; legacy migration from conf.discovery.plugins on first boot (idempotent, empty-table-guarded) - api_plugins.js (replaces routes/plugins.js): types/list/get/create/update/ secrets/test/load/unload/run/delete/runs; admin-gated; secrets always masked - /plugins page (plugins.ejs) + nav; Agents & Scheduler tab removed from /directory; /docs/agents aliased to /docs/plugins - proxmox/unifi/nmap gained manifests (configSchema/validate/run alias) - tests/plugins.test.js: registry unit + plugin_secrets (mocked bao-conf) + PluginInstance model round-trip/unique-slug - docs (plugins.md, vault.md, _config.yml, API.md) + 1.16.1 -> 1.17.0 Requires theta-suite >= v1.30.1 for the sso-broker secret/plugins/* grant; fails-soft with a clear error if absent. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,172 @@
|
||||
'use strict';
|
||||
|
||||
// Plugin type registry.
|
||||
//
|
||||
// A **plugin type** is a module under nodejs/plugins/<category>/<type>.js
|
||||
// exporting a manifest:
|
||||
//
|
||||
// { type, category, name, description, configSchema[], validate(), run() }
|
||||
//
|
||||
// `configSchema` is an array of field descriptors that drive the admin UI form
|
||||
// and API validation. Fields with `secret: true` are stored in OpenBao
|
||||
// (secret/plugins/<instance-id>/conf via utils/plugin_secrets.js); all other
|
||||
// field values live in the PluginInstance DB row's `config` JSON column.
|
||||
//
|
||||
// `run(cfg)` does the work; the discovery plugins keep their historical
|
||||
// `discover(cfg)` name and add `run` as an alias (the loader uses `run`).
|
||||
//
|
||||
// A **plugin instance** (models/plugin_instance.js) is a configured, loadable
|
||||
// copy of a type — you can have several of the same type. This registry only
|
||||
// knows about *types*; instances live in the DB.
|
||||
//
|
||||
// The scan happens once at require time (the set of installed .js files does
|
||||
// not change without a redeploy). Runtime load/unload is per-instance, not
|
||||
// per-type — adding a new plugin type still needs a restart.
|
||||
|
||||
const fs = require('fs');
|
||||
const path = require('path');
|
||||
|
||||
const pluginsRoot = path.join(__dirname, '../plugins');
|
||||
const MASK = '********';
|
||||
|
||||
// type -> module. Built once.
|
||||
const _modules = new Map();
|
||||
// type -> manifest summary (a safe, serializable subset for the UI/API).
|
||||
const _summaries = [];
|
||||
|
||||
function loadAll() {
|
||||
_modules.clear();
|
||||
_summaries.length = 0;
|
||||
if (!fs.existsSync(pluginsRoot)) return;
|
||||
for (const category of fs.readdirSync(pluginsRoot)) {
|
||||
const catDir = path.join(pluginsRoot, category);
|
||||
const stat = fs.statSync(catDir);
|
||||
if (!stat.isDirectory()) continue;
|
||||
for (const file of fs.readdirSync(catDir)) {
|
||||
if (!file.endsWith('.js')) continue;
|
||||
const type = path.basename(file, '.js');
|
||||
// require fresh-ish: a plugin file should be idempotent to load. Clear
|
||||
// from the cache so a future re-scan (e.g. in tests) picks up edits.
|
||||
const full = path.join(catDir, file);
|
||||
delete require.cache[require.resolve(full)];
|
||||
const mod = require(full);
|
||||
// Backfill manifest defaults so older plugins (only exporting discover)
|
||||
// still register with a usable summary.
|
||||
const manifest = {
|
||||
type: mod.type || type,
|
||||
category: mod.category || category,
|
||||
name: mod.name || type,
|
||||
description: mod.description || '',
|
||||
configSchema: Array.isArray(mod.configSchema) ? mod.configSchema : [],
|
||||
validate: typeof mod.validate === 'function' ? mod.validate : null,
|
||||
run: typeof mod.run === 'function' ? mod.run
|
||||
: typeof mod.discover === 'function' ? mod.discover : null
|
||||
};
|
||||
_modules.set(manifest.type, { mod, manifest });
|
||||
_summaries.push({
|
||||
type: manifest.type,
|
||||
category: manifest.category,
|
||||
name: manifest.name,
|
||||
description: manifest.description,
|
||||
configSchema: manifest.configSchema
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
loadAll();
|
||||
|
||||
// All registered plugin types, as serializable summaries (no functions).
|
||||
// Used by GET /api/plugins/types to build the "New Plugin" picker + form.
|
||||
function getTypes() {
|
||||
return _summaries.map(s => ({ ...s }));
|
||||
}
|
||||
|
||||
// The raw module for a type (has run/validate/discover). Throws if unknown.
|
||||
function getModule(type) {
|
||||
const entry = _modules.get(type);
|
||||
if (!entry) {
|
||||
const err = new Error(`Unknown plugin type: ${type}`);
|
||||
err.status = 400;
|
||||
throw err;
|
||||
}
|
||||
return entry.mod;
|
||||
}
|
||||
|
||||
// The manifest summary for a type. Returns null if unknown (callers gate on
|
||||
// this to validate a pluginType before creating an instance).
|
||||
function getManifest(type) {
|
||||
const entry = _modules.get(type);
|
||||
return entry ? entry.manifest : null;
|
||||
}
|
||||
|
||||
// Keys of the secret fields in a type's configSchema.
|
||||
function secretKeys(type) {
|
||||
const m = getManifest(type);
|
||||
if (!m) return [];
|
||||
return m.configSchema.filter(f => f.secret).map(f => f.key);
|
||||
}
|
||||
|
||||
// Non-secret field keys in a type's configSchema.
|
||||
function publicKeys(type) {
|
||||
const m = getManifest(type);
|
||||
if (!m) return [];
|
||||
return m.configSchema.filter(f => !f.secret).map(f => f.key);
|
||||
}
|
||||
|
||||
// All declared field keys (secret + non-secret) — for required-field validation.
|
||||
function fieldKeys(type) {
|
||||
const m = getManifest(type);
|
||||
if (!m) return [];
|
||||
return m.configSchema.map(f => f.key);
|
||||
}
|
||||
|
||||
// Required field keys.
|
||||
function requiredKeys(type) {
|
||||
const m = getManifest(type);
|
||||
if (!m) return [];
|
||||
return m.configSchema.filter(f => f.required).map(f => f.key);
|
||||
}
|
||||
|
||||
// Replace each present secret value with MASK, keeping the keys so the UI can
|
||||
// render a prefilled (masked) password field. Non-secret values are passed
|
||||
// through unchanged. `values` is a plain object of field->value.
|
||||
function mask(type, values) {
|
||||
if (!values || typeof values !== 'object') return values;
|
||||
const sk = new Set(secretKeys(type));
|
||||
const out = {};
|
||||
for (const [k, v] of Object.entries(values)) {
|
||||
out[k] = sk.has(k) && v ? MASK : v;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
// Split a flat {field: value} object (as the UI/API sends it) into non-secret
|
||||
// config (for the DB row) and secret values (for OpenBao). Unknown keys are
|
||||
// dropped — only declared configSchema fields are kept.
|
||||
function splitConfig(type, flat) {
|
||||
const manifest = getManifest(type);
|
||||
const config = {};
|
||||
const secrets = {};
|
||||
if (!manifest || !flat) return { config, secrets };
|
||||
for (const f of manifest.configSchema) {
|
||||
if (!(f.key in flat)) continue;
|
||||
if (f.secret) secrets[f.key] = flat[f.key];
|
||||
else config[f.key] = flat[f.key];
|
||||
}
|
||||
return { config, secrets };
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
getTypes,
|
||||
getModule,
|
||||
getManifest,
|
||||
secretKeys,
|
||||
publicKeys,
|
||||
fieldKeys,
|
||||
requiredKeys,
|
||||
mask,
|
||||
splitConfig,
|
||||
// for tests
|
||||
_reload: loadAll
|
||||
};
|
||||
+181
-59
@@ -1,5 +1,27 @@
|
||||
'use strict';
|
||||
|
||||
// Discovery / plugin scheduler.
|
||||
//
|
||||
// Generalized from the one-shot discovery-plugin loader: plugin *types* live
|
||||
// under nodejs/plugins/<category>/<type>.js (see services/plugin_registry.js),
|
||||
// and configured, loadable/unloadable *instances* live in the PluginInstance
|
||||
// table (models/plugin_instance.js). This module schedules enabled instances
|
||||
// on cron via BullMQ JobSchedulers and runs them in a Worker.
|
||||
//
|
||||
// Each instance owns a stable JobScheduler id (`plugin:<instanceId>`) so load/
|
||||
// unload can add/remove a single schedule without disturbing the others —
|
||||
// `upsertJobScheduler`/`removeJobScheduler` (BullMQ v6) take that id directly.
|
||||
//
|
||||
// Per-instance secrets are merged in from OpenBao (utils/plugin_secrets.js) at
|
||||
// run time; the plugin's run()/discover() receives the combined non-secret
|
||||
// config + secret values as a single `config` object, exactly as the legacy
|
||||
// static-config path did.
|
||||
|
||||
const { Queue, Worker } = require('bullmq');
|
||||
const { DiscoveryReconciler } = require('./discovery_reconciler');
|
||||
const pluginRegistry = require('./plugin_registry');
|
||||
const pluginSecrets = require('../utils/plugin_secrets');
|
||||
const { PluginInstance, STATUS } = require('../models/plugin_instance');
|
||||
const Redis = require('ioredis');
|
||||
|
||||
// Ensure Redis connection works for BullMQ
|
||||
@@ -8,80 +30,180 @@ const connection = new Redis(process.env.REDIS_URL || 'redis://127.0.0.1:6379',
|
||||
|
||||
const discoveryQueue = new Queue('discovery', { connection });
|
||||
|
||||
// Load plugins
|
||||
const fs = require('fs');
|
||||
const path = require('path');
|
||||
const pluginsDir = path.join(__dirname, '../plugins/discovery');
|
||||
|
||||
let plugins = {};
|
||||
|
||||
if (fs.existsSync(pluginsDir)) {
|
||||
fs.readdirSync(pluginsDir).forEach(file => {
|
||||
if (file.endsWith('.js')) {
|
||||
const name = path.basename(file, '.js');
|
||||
plugins[name] = require(path.join(pluginsDir, file));
|
||||
}
|
||||
});
|
||||
}
|
||||
const RUN = 'run_plugin';
|
||||
const GC = 'garbage_collect';
|
||||
function pluginSchedulerId(id) { return `plugin:${id}`; }
|
||||
|
||||
const worker = new Worker('discovery', async job => {
|
||||
if (job.name === 'run_plugin') {
|
||||
const { pluginName, config } = job.data;
|
||||
if (plugins[pluginName]) {
|
||||
console.log(`[Scheduler] Running plugin: ${pluginName}`);
|
||||
try {
|
||||
const payload = await plugins[pluginName].discover(config);
|
||||
await DiscoveryReconciler.reconcile(pluginName, payload);
|
||||
} catch (err) {
|
||||
console.error(`[Scheduler] Plugin ${pluginName} failed:`, err);
|
||||
}
|
||||
}
|
||||
} else if (job.name === 'garbage_collect') {
|
||||
console.log(`[Scheduler] Running garbage collection`);
|
||||
if (job.name === RUN) {
|
||||
await runPluginJob(job.data && job.data.instanceId);
|
||||
} else if (job.name === GC) {
|
||||
console.log('[Scheduler] Running garbage collection');
|
||||
await DiscoveryReconciler.garbageCollect();
|
||||
}
|
||||
}, { connection });
|
||||
|
||||
// Function to start scheduling
|
||||
async function initScheduler(discoveryConfig) {
|
||||
// Clear old repeatable jobs (BullMQ v6 uses JobSchedulers)
|
||||
try {
|
||||
const schedulers = await discoveryQueue.getJobSchedulers();
|
||||
for (const job of schedulers) {
|
||||
await discoveryQueue.removeJobScheduler(job.id);
|
||||
}
|
||||
} catch (e) {
|
||||
console.log('[Scheduler] Could not clear old job schedulers (may not be supported or none exist)');
|
||||
// Run one plugin instance. Loads the row (skip silently if it was deleted or
|
||||
// disabled after the job was enqueued), merges its OpenBao secrets into its
|
||||
// config, calls the plugin's run()/discover(), and — for discovery plugins —
|
||||
// reconciles the result into the resource graph under the instance's slug.
|
||||
// Bookkeeping (lastRunAt/lastStatus/lastError) is stamped on the row so the UI
|
||||
// can show run state without querying BullMQ.
|
||||
async function runPluginJob(instanceId) {
|
||||
if (!instanceId) { console.warn('[Scheduler] run_plugin job with no instanceId'); return; }
|
||||
const instance = await PluginInstance.get(instanceId);
|
||||
if (!instance) { console.warn(`[Scheduler] instance ${instanceId} gone — skipping`); return; }
|
||||
if (!instance.enabled) { console.warn(`[Scheduler] instance ${instance.slug} (${instanceId}) disabled — skipping`); return; }
|
||||
|
||||
let mod;
|
||||
try { mod = pluginRegistry.getModule(instance.pluginType); }
|
||||
catch (err) {
|
||||
console.error(`[Scheduler] instance ${instance.slug}: type ${instance.pluginType} unavailable:`, err.message);
|
||||
await instance.update({ lastRunAt: Date.now(), lastStatus: STATUS.ERROR, lastError: `plugin type unavailable: ${instance.pluginType}` });
|
||||
return;
|
||||
}
|
||||
|
||||
// Schedule Garbage Collection
|
||||
await discoveryQueue.add('garbage_collect', {}, { repeat: { pattern: '0 0 * * *' } }); // Daily
|
||||
const runFn = mod.run || mod.discover;
|
||||
if (typeof runFn !== 'function') {
|
||||
console.error(`[Scheduler] instance ${instance.slug}: type ${instance.pluginType} has no run()/discover()`);
|
||||
await instance.update({ lastRunAt: Date.now(), lastStatus: STATUS.ERROR, lastError: 'plugin type has no run()/discover()' });
|
||||
return;
|
||||
}
|
||||
|
||||
// Load plugin overrides from Redis
|
||||
let overrides = {};
|
||||
console.log(`[Scheduler] Running plugin: ${instance.slug} (${instance.pluginType})`);
|
||||
await instance.update({ lastRunAt: Date.now(), lastStatus: STATUS.RUNNING, lastError: null });
|
||||
try {
|
||||
const data = await connection.hgetall('discovery_plugins');
|
||||
for (const [k, v] of Object.entries(data)) {
|
||||
overrides[k] = JSON.parse(v);
|
||||
const cfg = await pluginSecrets.mergeForRun(instance);
|
||||
const payload = await runFn(cfg);
|
||||
if (instance.category === 'discovery') {
|
||||
await DiscoveryReconciler.reconcile(instance.slug, payload);
|
||||
}
|
||||
await instance.update({ lastStatus: STATUS.OK, lastError: null });
|
||||
} catch (err) {
|
||||
console.error('[Scheduler] Failed to load plugin overrides from Redis', err);
|
||||
console.error(`[Scheduler] Plugin ${instance.slug} failed:`, err.message);
|
||||
await instance.update({ lastStatus: STATUS.ERROR, lastError: String(err.message || err) });
|
||||
}
|
||||
}
|
||||
|
||||
// Schedule Plugins based on config + overrides
|
||||
if (discoveryConfig && discoveryConfig.plugins) {
|
||||
for (const [name, config] of Object.entries(discoveryConfig.plugins)) {
|
||||
const mergedConfig = { ...config, ...(overrides[name] || {}) };
|
||||
if (mergedConfig.enabled && plugins[name]) {
|
||||
const cron = mergedConfig.cron || '0 * * * *'; // Default hourly
|
||||
await discoveryQueue.add('run_plugin', { pluginName: name, config: mergedConfig }, { repeat: { pattern: cron } });
|
||||
console.log(`[Scheduler] Scheduled plugin ${name} with cron ${cron}`);
|
||||
|
||||
// Also run once immediately
|
||||
await discoveryQueue.add('run_plugin', { pluginName: name, config: mergedConfig });
|
||||
}
|
||||
// Schedule one instance: upsert a repeatable JobScheduler keyed by its id. Does
|
||||
// NOT trigger an immediate run — call runInstanceNow(id) separately for that
|
||||
// (used on boot and on "load"). Safe to call repeatedly (upsert is idempotent
|
||||
// and will update the cron if it changed).
|
||||
async function scheduleInstance(instance) {
|
||||
if (!instance || !instance.id) return;
|
||||
if (!instance.enabled) { await unscheduleInstance(instance.id); return; }
|
||||
const cron = instance.cron || '0 * * * *';
|
||||
await discoveryQueue.upsertJobScheduler(pluginSchedulerId(instance.id), { pattern: cron }, {
|
||||
name: RUN,
|
||||
data: { instanceId: instance.id }
|
||||
});
|
||||
console.log(`[Scheduler] Scheduled instance ${instance.slug} with cron ${cron}`);
|
||||
}
|
||||
|
||||
// Remove an instance's repeatable schedule. No-op if it had none.
|
||||
async function unscheduleInstance(id) {
|
||||
if (!id) return;
|
||||
try { await discoveryQueue.removeJobScheduler(pluginSchedulerId(id)); }
|
||||
catch (err) { /* missing scheduler is fine */ }
|
||||
}
|
||||
|
||||
// Enqueue a single immediate run for an instance (the "Run now" button / boot
|
||||
// kick). Runs once regardless of enabled, on top of any schedule.
|
||||
async function runInstanceNow(id) {
|
||||
if (!id) return;
|
||||
await discoveryQueue.add(RUN, { instanceId: id });
|
||||
}
|
||||
|
||||
// One-time legacy migration: if the PluginInstance table is empty AND
|
||||
// conf.discovery.plugins has entries (the old static-config shape), seed one
|
||||
// instance per configured type and copy its secret fields into OpenBao. After
|
||||
// the first boot, the table is non-empty and the static config is ignored.
|
||||
// Idempotent (guarded by the empty-table check).
|
||||
async function migrateLegacyPlugins(discoveryConfig) {
|
||||
const existing = await PluginInstance.list();
|
||||
if (existing && existing.length) return;
|
||||
|
||||
const legacy = discoveryConfig && discoveryConfig.plugins;
|
||||
if (!legacy || typeof legacy !== 'object') return;
|
||||
const names = Object.keys(legacy);
|
||||
if (!names.length) return;
|
||||
|
||||
console.log(`[Scheduler] Migrating ${names.length} legacy discovery plugin(s) to instances…`);
|
||||
for (const name of names) {
|
||||
const entry = legacy[name] || {};
|
||||
const manifest = pluginRegistry.getManifest(name);
|
||||
if (!manifest) {
|
||||
console.warn(`[Scheduler] legacy plugin '${name}' has no registered type — skipping`);
|
||||
continue;
|
||||
}
|
||||
// splitConfig keeps only declared configSchema fields and separates secret
|
||||
// from non-secret. Legacy `enabled`/`cron` are not in configSchema, so they
|
||||
// are dropped here and read from the entry directly below.
|
||||
const { config, secrets } = pluginRegistry.splitConfig(name, entry);
|
||||
const instance = await PluginInstance.create({
|
||||
pluginType: name,
|
||||
category: manifest.category,
|
||||
name: manifest.name,
|
||||
slug: name,
|
||||
enabled: entry.enabled !== false,
|
||||
cron: entry.cron || '0 * * * *',
|
||||
config,
|
||||
created_by: 'legacy-migration'
|
||||
});
|
||||
try {
|
||||
await pluginSecrets.write(instance.id, secrets);
|
||||
console.log(`[Scheduler] migrated '${name}' -> instance ${instance.id} (slug ${instance.slug})`);
|
||||
} catch (err) {
|
||||
// The instance row exists; if we can't write secrets (e.g. the sso-broker
|
||||
// policy predates theta-suite v1.30.1) the operator gets a clear error
|
||||
// from the API on edit, and the instance still runs with its non-secret
|
||||
// config. Don't delete the row — the operator just needs to re-run
|
||||
// setup.sh and edit/save the secrets.
|
||||
console.error(`[Scheduler] migrated '${name}' row but FAILED to write secrets:`, err.message);
|
||||
await instance.update({ lastStatus: STATUS.ERROR, lastError: `secret migration failed: ${err.message}` });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { initScheduler, discoveryQueue, connection };
|
||||
// Boot-time initialization: clear stale schedulers, schedule garbage collection,
|
||||
// migrate any legacy static-config plugins, then schedule every enabled
|
||||
// instance and kick one immediate run for each.
|
||||
async function initScheduler(discoveryConfig) {
|
||||
// Clear stale plugin/gc schedulers from a previous boot. Other-named
|
||||
// schedulers (none in this app) are left alone.
|
||||
try {
|
||||
const schedulers = await discoveryQueue.getJobSchedulers();
|
||||
for (const s of schedulers) {
|
||||
if (s.name === RUN || s.name === GC) {
|
||||
await discoveryQueue.removeJobScheduler(s.key || s.id);
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
console.log('[Scheduler] Could not clear old job schedulers:', e.message);
|
||||
}
|
||||
|
||||
// Daily garbage collection of stale discovery resources.
|
||||
await discoveryQueue.upsertJobScheduler(GC, { pattern: '0 0 * * *' }, { name: GC, data: {} });
|
||||
|
||||
try {
|
||||
await migrateLegacyPlugins(discoveryConfig);
|
||||
} catch (err) {
|
||||
console.error('[Scheduler] legacy migration failed:', err.message);
|
||||
}
|
||||
|
||||
const enabled = await PluginInstance.listEnabled();
|
||||
for (const instance of enabled) {
|
||||
await scheduleInstance(instance);
|
||||
await runInstanceNow(instance.id); // boot kick
|
||||
}
|
||||
console.log(`[Scheduler] initialized — ${enabled.length} instance(s) scheduled`);
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
initScheduler,
|
||||
scheduleInstance,
|
||||
unscheduleInstance,
|
||||
runInstanceNow,
|
||||
discoveryQueue,
|
||||
connection
|
||||
};
|
||||
Reference in New Issue
Block a user