Spaces:
Runtime error
Runtime error
File size: 67,800 Bytes
46252cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 | import { Injectable, OnApplicationBootstrap, OnModuleInit, OnModuleDestroy, Optional } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { ModuleRef } from '@nestjs/core';
import { toNeutralJid, userPart } from '../../engine/identity/wa-id';
import { LidMappingStoreService } from '../../engine/identity/lid-mapping-store.service';
import { AsyncLocalStorage } from 'async_hooks';
import * as fs from 'fs';
import * as path from 'path';
import { createLogger } from '../../common/services/logger.service';
import { HookManager, HookEvent, KNOWN_HOOK_EVENTS, isKnownHookEvent } from '../hooks';
import {
PluginCapabilityError,
PluginCapabilityPermission,
PluginEngineReadCapability,
PluginManifest,
PluginMessagingCapability,
PluginNetCapability,
PluginConversationsCapability,
PluginHandoverCapability,
PluginMappingsCapability,
PluginInstance,
PluginStatus,
PluginContext,
IPlugin,
PluginType,
PluginLogger,
PluginConfigSchema,
validateIngressManifest,
warnUnauthenticatedIngressRoutes,
} from './plugin.interfaces';
import { effectiveNetAllow, isNetHostAllowed, performPluginFetch } from './plugin-net';
import { PluginStorageService } from './plugin-storage.service';
import { isPluginActiveForSession, resolvePluginConfig } from './plugin-activation';
import { PluginWorkerHost } from './sandbox/plugin-worker-host';
import { WorkerThreadChannel } from './sandbox/worker-thread-channel';
import { dispatchCapabilityVerb } from './sandbox/capability-router';
import { PluginLogLevel } from './sandbox/protocol';
import { buildConversationSendFacade, ConversationMediaType } from './conversation-send-facade';
import { shouldDispatchToPlugin } from './handover-gate';
import { makeOnWebhookSubscribe } from './webhook-subscribe.util';
import { registerPluginSearchProvider, unregisterPluginSearchProvider } from './search-provider-registration.util';
import { INGRESS_DISPATCH_TIMEOUT_MS } from '../../modules/integration/integration.constants';
import type { MessageService } from '../../modules/message/message.service';
import type { SessionService } from '../../modules/session/session.service';
import type { IWhatsAppEngine } from '../../engine/interfaces/whatsapp-engine.interface';
import type { ConversationMappingService } from '../../modules/integration/conversation-mapping.service';
import type { PluginInstanceService } from '../../modules/integration/plugin-instance.service';
import type { IngressJobData } from '../../modules/queue/processors/ingress.processor';
import type { SearchProviderRegistry } from '../../modules/search/search-provider.registry';
/** Default per-plugin heap cap for the sandbox worker; an OOM terminates the worker, not the host. */
const SANDBOX_MAX_OLD_GEN_MB = 256;
/** Time budget for a sandboxed plugin's hook handler before the chain proceeds without it. */
const SANDBOX_HOOK_TIMEOUT_MS = 5000;
/** A sandboxed plugin's healthCheck must answer within this, else it's reported unhealthy (not hung). */
const SANDBOX_HEALTH_TIMEOUT_MS = 5000;
/** A sandboxed plugin's search handler must answer within this, else /search fails fast (not hung). */
const SANDBOX_SEARCH_TIMEOUT_MS = 10000;
/**
* A sandboxed plugin's load()/onLoad/onEnable/onDisable must complete within this, else the worker is
* torn down and the operation fails β a wedged lifecycle can't hang the enable/disable request (and
* the ADMIN HTTP call behind it) forever. Generous on purpose: a slow-but-valid onEnable that opens
* connections should still finish well under it.
*/
const SANDBOX_LIFECYCLE_TIMEOUT_MS = 30000;
/**
* Max concurrent worker-initiated capability calls per sandboxed plugin. A burst beyond this is rejected
* (the plugin sees a thrown Error) rather than amplified into unbounded host-side sends/fetches/writes.
*/
const SANDBOX_MAX_INFLIGHT_CAPS = 32;
/**
* Host process.env keys an untrusted plugin worker is allowed to see. Everything else β secrets like
* API_MASTER_KEY, API_KEY_PEPPER, the DATABASE_/REDIS_ vars, DOCKER_HOST β is withheld. The worker is
* a thread, so it needs no PATH to start and require() resolves via module paths, not env.
*/
const SANDBOX_ENV_ALLOWLIST = ['NODE_ENV', 'NODE_EXTRA_CA_CERTS', 'TZ'] as const;
/**
* Resolve a plugin's `main` entry to an absolute path, asserting it stays inside
* <pluginsDir>/<pluginId>. `main` comes from a user-supplied manifest, so a
* value like '../../etc/passwd' (or an absolute path) must be rejected BEFORE require().
*/
export function resolvePluginMainPath(pluginsDir: string, pluginId: string, main: string): string {
const base = path.resolve(pluginsDir, pluginId);
const mainPath = path.resolve(base, main);
if (mainPath !== base && !mainPath.startsWith(base + path.sep)) {
throw new Error(`Plugin ${pluginId} main path escapes the plugin directory`);
}
return mainPath;
}
/**
* Build the minimal, allowlisted env for an untrusted plugin worker so it never inherits host secrets.
* Only {@link SANDBOX_ENV_ALLOWLIST} keys are forwarded (unset keys are omitted, not emitted as
* `undefined`), and NODE_ENV defaults to 'production' when the host has none.
*/
export function buildSandboxWorkerEnv(source: NodeJS.ProcessEnv = process.env): NodeJS.ProcessEnv {
const env: NodeJS.ProcessEnv = {};
for (const key of SANDBOX_ENV_ALLOWLIST) {
if (source[key] !== undefined) env[key] = source[key];
}
env.NODE_ENV = source.NODE_ENV ?? 'production';
return env;
}
/**
* Translate a normalized conversation media send into the concrete MessageService media method for the
* envelope's type. Kept pure (no `this`) so the loader binds it directly and it can be unit-tested in
* isolation. The switch is exhaustive over ConversationMediaType β adding a type without a case is a
* compile error here rather than a silent runtime fall-through.
*/
export function dispatchConversationMedia(
svc: Pick<MessageService, 'sendImage' | 'sendVideo' | 'sendAudio' | 'sendDocument'>,
sessionId: string,
opts: { chatId: string; url: string; type: ConversationMediaType; caption?: string },
): Promise<unknown> {
const dto = { chatId: opts.chatId, url: opts.url, caption: opts.caption };
switch (opts.type) {
case 'image':
return svc.sendImage(sessionId, dto);
case 'video':
return svc.sendVideo(sessionId, dto);
case 'audio':
return svc.sendAudio(sessionId, dto);
case 'voice':
// A voice envelope is a PTT note: sendAudio with ptt classifies it as 'voice' and defaults the
// codec to audio/ogg;opus, so it renders as a WhatsApp voice bubble rather than an audio file.
return svc.sendAudio(sessionId, { ...dto, ptt: true });
case 'file':
return svc.sendDocument(sessionId, dto);
}
}
// Plugin ids whose bundled-extension code was permanently removed (v0.7 β superseded by the
// marketplace chat-flow / group-translate; also reserved in plugin-installer). A leftover
// directory without a manifest marks them as deleted on disk, so the stale registry entry (which
// still reports them installed/enabled) is pruned on boot. Scoped to these known ids so a
// temporarily-unreadable plugin dir (e.g. an unmounted volume) never loses its persisted config.
const LEGACY_REMOVED_PLUGIN_IDS = new Set(['auto-reply', 'translation']);
/**
* Fill config keys the schema declares a `default` for and that are absent (undefined) in the
* stored config. Seeding happens at LOAD time (fresh installs and every boot), so a plugin whose
* schema fields carry defaults never runs its lifecycle with them missing β the failure class of
* "enable throws: <field> is required/has no value" for defaulted fields. Explicit values β even
* null β are never overwritten, and object/array defaults are deep-cloned so the seeded runtime
* config and the persisted entry can't share a mutable reference. Required fields WITHOUT a
* declared default stay absent on purpose: those need real operator input, not an invented value.
*/
export function seedConfigDefaults(
schema: PluginConfigSchema | undefined,
config: Record<string, unknown>,
): Record<string, unknown> {
const properties = schema?.properties;
if (!properties) return config;
let seeded: Record<string, unknown> | undefined;
for (const [key, field] of Object.entries(properties)) {
if (config[key] !== undefined || field === null || typeof field !== 'object') continue;
const value = field.default;
if (value === undefined) continue;
if (!seeded) seeded = { ...config };
seeded[key] = value !== null && typeof value === 'object' ? structuredClone(value) : value;
}
return seeded ?? config;
}
@Injectable()
export class PluginLoaderService implements OnModuleInit, OnApplicationBootstrap, OnModuleDestroy {
private readonly logger = createLogger('PluginLoaderService');
private readonly plugins = new Map<string, PluginInstance>();
/** Plugin ids whose enable() is in flight β a synchronous lock so concurrent enables can't double-run. */
private readonly enabling = new Set<string>();
// Live worker host per enabled sandboxed (untrusted) plugin. Built-ins are not in here.
private readonly sandboxHosts = new Map<string, PluginWorkerHost>();
// Carries the firing event's sessionId across an in-process hook handler so ctx.config (a getter)
// resolves the per-session slice. Per async call tree, so concurrent sessions don't cross over.
private readonly hookSession = new AsyncLocalStorage<{ sessionId?: string }>();
private readonly pluginsDir: string;
constructor(
private readonly configService: ConfigService,
private readonly hookManager: HookManager,
private readonly pluginStorage: PluginStorageService,
// Resolves MessageService/SessionService lazily inside capability verbs. ModuleRef is used
// instead of constructor injection to avoid the provider cycle
// PluginLoaderService -> SessionService -> EngineFactory -> PluginLoaderService.
private readonly moduleRef: ModuleRef,
// Shared lid->phone table (EngineModule is @Global and exports it). Optional so the many unit tests
// that construct this service with the 4 prior args still compile; when absent, canonicalChatId
// degrades to identity (no @lid resolution).
@Optional() private readonly lidMappingStore?: LidMappingStoreService,
) {
this.pluginsDir = this.configService.get<string>('plugins.dir') ?? './plugins';
}
onModuleInit(): void {
// Load built-in plugins first (synchronous registration)
this.loadBuiltInPlugins();
// Then load user plugins if directory exists
if (fs.existsSync(this.pluginsDir)) {
this.loadPluginsFromDirectory(this.pluginsDir);
}
this.logger.log(`Loaded ${this.plugins.size} plugins`, {
action: 'plugins_loaded',
count: this.plugins.size,
});
}
/**
* Re-enable the plugins the operator had enabled (#856). `status` cannot carry that across a restart
* β it describes the runtime, and loading never runs a plugin β so the decision is read from the
* separately persisted `enabledByOperator`. Without this, every restart (an upgrade, a host reboot, a
* Docker restart policy) silently switched off every extension, and a relay simply stopped relaying.
*
* Runs at bootstrap rather than in onModuleInit so the rest of the app is wired before any plugin
* code executes. Built-ins are skipped: an engine is enabled by EngineFactory against the configured
* engine.type, and enabling a non-active engine here would be rejected anyway.
*
* Best-effort and sequential, like the shutdown teardown: a plugin that cannot come back is logged
* and left in ERROR, and never holds up the gateway.
*/
async onApplicationBootstrap(): Promise<void> {
const restorable = this.getAllPlugins().filter(
p => !p.builtIn && this.pluginStorage.getPluginEntry(p.manifest.id)?.enabledByOperator === true,
);
for (const plugin of restorable) {
const pluginId = plugin.manifest.id;
try {
await this.enablePlugin(pluginId);
} catch (error) {
this.logger.error(
`Failed to restore plugin ${pluginId} on startup; it stays disabled until re-enabled`,
error instanceof Error ? error.message : String(error),
{ pluginId, action: 'plugin_restore_failed' },
);
}
}
}
/**
* Graceful shutdown (SIGTERM β app.close()): run onDisable for every enabled plugin so it can flush
* buffers, close connections, and persist state. Previously onDisable only ran via the REST disable
* and uninstall paths, so a normal restart/deploy/scale-down skipped it and stateful plugins lost
* in-flight work. Best-effort and sequential: one plugin's failure must not block the others.
*/
async onModuleDestroy(): Promise<void> {
const enabled = this.getAllPlugins().filter(p => p.status === PluginStatus.ENABLED);
for (const plugin of enabled) {
try {
await this.disablePlugin(plugin.manifest.id);
} catch (error) {
this.logger.error(
`Failed to disable plugin ${plugin.manifest.id} during shutdown`,
error instanceof Error ? error.message : String(error),
{ pluginId: plugin.manifest.id, action: 'plugin_shutdown_disable_failed' },
);
}
}
}
private loadBuiltInPlugins(): void {
// Built-in plugins are registered programmatically
// This will be used by Phase 4 to register engine plugins
this.logger.debug('Built-in plugins loading point (Phase 4)', {
action: 'builtin_plugins_init',
});
}
private loadPluginsFromDirectory(dir: string): void {
const entries = fs.readdirSync(dir, { withFileTypes: true });
for (const entry of entries) {
// Skip non-directories and dot-prefixed dirs (e.g. a crash-leftover `.<id>.bak` update backup),
// so a half-finished update can't be re-loaded as a duplicate-id plugin on the next boot.
if (!entry.isDirectory() || entry.name.startsWith('.')) continue;
const pluginPath = path.join(dir, entry.name);
const manifestPath = path.join(pluginPath, 'manifest.json');
if (!fs.existsSync(manifestPath)) {
this.logger.warn(`Plugin ${entry.name} missing manifest.json`, {
pluginPath,
action: 'manifest_missing',
});
if (LEGACY_REMOVED_PLUGIN_IDS.has(entry.name)) {
this.pluginStorage.deletePluginEntry(entry.name);
this.logger.log(`Pruned stale registry entry for removed built-in plugin: ${entry.name}`, {
action: 'registry_ghost_pruned',
});
}
continue;
}
try {
this.loadPlugin(pluginPath);
} catch (error) {
this.logger.error(
`Failed to load plugin ${entry.name}`,
error instanceof Error ? error.message : String(error),
{ pluginPath, action: 'plugin_load_failed' },
);
}
}
}
loadPlugin(pluginPath: string): PluginInstance {
const manifestPath = path.join(pluginPath, 'manifest.json');
const manifestContent = fs.readFileSync(manifestPath, 'utf-8');
const manifest = JSON.parse(manifestContent) as PluginManifest;
// Validate manifest
if (!manifest.id || !manifest.name || !manifest.version || !manifest.type || !manifest.main) {
throw new Error(`Invalid manifest: missing required fields`);
}
// Reject a malformed ingress declaration (SDK-major mismatch, missing webhook:ingress permission,
// duplicate/empty routes, non-positive toleranceSec) at load time instead of letting it silently
// load and become provisionable. No-op for plugins that declare no ingress.
validateIngressManifest(manifest);
// Surface a loud warning for any ingress route that skips signature verification β a scheme:'none'
// route is a fully-unauthenticated public endpoint that can trigger WhatsApp sends. Additive (a
// warning, not a refusal) so a legit scheme:'none' deployment still boots.
warnUnauthenticatedIngressRoutes(manifest, this.logger);
// Check if plugin already loaded
if (this.plugins.has(manifest.id)) {
throw new Error(`Plugin ${manifest.id} is already loaded`);
}
// Load any persisted config + per-session activation + per-session config so an operator's choices
// survive a restart.
const storedConfig = this.pluginStorage.getPluginConfig(manifest.id) ?? {};
const storedSessions = this.pluginStorage.getPluginSessions(manifest.id) ?? undefined;
const storedSessionConfig = this.pluginStorage.getPluginSessionConfig(manifest.id) ?? undefined;
const pluginInstance: PluginInstance = {
manifest,
status: PluginStatus.INSTALLED,
// Seed schema-declared defaults under the stored config, so a defaulted field is never
// missing when the plugin later runs (explicit values are never overwritten).
config: seedConfigDefaults(manifest.configSchema, storedConfig),
instance: null,
loadedAt: new Date(),
builtIn: false,
activeSessions: storedSessions,
sessionConfig: storedSessionConfig,
};
this.plugins.set(manifest.id, pluginInstance);
// Ensure a registry entry exists so later enable/disable/config writes persist.
this.ensureRegistryEntry(manifest, false);
this.logger.log(`Plugin loaded: ${manifest.name} v${manifest.version}`, {
pluginId: manifest.id,
type: manifest.type,
action: 'plugin_loaded',
});
return pluginInstance;
}
/**
* Ensure a freshly-loaded plugin has a persisted registry entry, so later enable/disable/config
* writes (which only update an EXISTING entry) actually persist instead of silently no-op'ing.
* Creates a complete INSTALLED entry when none exists; an existing entry's persisted status/config
* is left untouched. Best-effort (saveRegistry swallows fs errors, so a disk failure never turns a
* load into a 500). Does NOT enable or run the plugin β boot never auto-executes plugin code.
*/
private ensureRegistryEntry(manifest: PluginManifest, builtIn: boolean): void {
// Reconcile the persisted entry with the freshly-loaded runtime: loading never runs the plugin, so
// the entry's status is (re)set to INSTALLED to match the runtime. Enabling is a separate step that
// runs the lifecycle β at bootstrap for a plugin the operator had enabled (see
// onApplicationBootstrap), or on an explicit ADMIN action. The operator's persisted config and
// enable decision are preserved so settings/secrets and the decision itself survive. Best-effort:
// saveRegistry swallows fs errors, so a disk failure never turns a load into a 500.
const existing = this.pluginStorage.getPluginEntry(manifest.id);
// The operator's standing enable decision (#856). `status` below is deliberately reset, so intent
// has to live in its own field or a restart loses it. A pre-#856 row has no such field: adopt it
// from a status of ENABLED, which can only have been written by an explicit enable since the last
// boot (every boot rewrites the status to INSTALLED), so it is a faithful record of the intent.
const enabledByOperator = existing?.enabledByOperator ?? existing?.status === PluginStatus.ENABLED;
this.pluginStorage.setPluginEntry({
id: manifest.id,
type: manifest.type,
name: manifest.name,
version: manifest.version,
status: PluginStatus.INSTALLED,
// The operator's persisted config survives, with schema-declared defaults seeded under it so
// the persisted entry matches the seeded runtime config (see loadPlugin).
config: seedConfigDefaults(manifest.configSchema, existing?.config ?? {}),
builtIn,
installedAt: existing?.installedAt ?? new Date(),
updatedAt: new Date(),
// setPluginEntry REPLACES the entry, so the operator's per-session activation + config must be
// carried over or every boot wipes them from disk (lost after the second restart).
activeSessions: existing?.activeSessions,
sessionConfig: existing?.sessionConfig,
enabledByOperator,
});
}
/**
* Record that the operator wants this plugin on (or off), so bootstrap can restore it (#856).
*
* Call this ONLY from an operator-facing action. In particular it must never be called from
* disablePlugin: onModuleDestroy disables every running plugin during a graceful shutdown, and
* treating that as "the operator turned it off" would erase the decision on the way out β which is
* the very bug this exists to fix, just moved somewhere harder to see.
*/
setOperatorEnabled(pluginId: string, enabled: boolean): void {
this.pluginStorage.setPluginEnabledByOperator(pluginId, enabled);
}
async enablePlugin(pluginId: string): Promise<void> {
const plugin = this.plugins.get(pluginId);
if (!plugin) {
throw new Error(`Plugin ${pluginId} not found`);
}
if (plugin.status === PluginStatus.ENABLED) {
return; // Already enabled
}
// Engines are mutually exclusive and pinned to the deployment's engine.type config (the factory
// reads that, not plugin status). Enabling a second engine at runtime would show two "active"
// engines and desync the factory, so reject anything but the configured active engine.
if (plugin.manifest.type === PluginType.ENGINE) {
const activeEngine = this.configService.get<string>('engine.type') ?? 'whatsapp-web.js';
if (pluginId !== activeEngine) {
throw new Error(
`Engine "${pluginId}" is not the active engine ("${activeEngine}"). Set engine.type and restart to switch engines.`,
);
}
}
// Concurrency guard: status flips to ENABLED only AFTER the awaits below, so two concurrent enable
// calls would both pass the check above, both run onEnable, and both register the plugin's hooks
// (duplicate side effects). Claim the enable synchronously here so a racing caller is rejected
// before any await; released in finally.
if (this.enabling.has(pluginId)) {
throw new Error(`Plugin ${pluginId} is already being enabled`);
}
this.enabling.add(pluginId);
try {
if (plugin.builtIn === false) {
await this.enableSandboxed(pluginId, plugin);
} else {
await this.enableInProcess(pluginId, plugin);
}
plugin.status = PluginStatus.ENABLED;
plugin.enabledAt = new Date();
plugin.error = undefined;
// Persist status
this.pluginStorage.setPluginStatus(pluginId, PluginStatus.ENABLED);
this.logger.log(`Plugin enabled: ${plugin.manifest.name}`, {
pluginId,
action: 'plugin_enabled',
});
} catch (error) {
plugin.status = PluginStatus.ERROR;
plugin.error = error instanceof Error ? error.message : String(error);
this.pluginStorage.setPluginStatus(pluginId, PluginStatus.ERROR);
// A plugin that subscribed hooks before its onLoad/onEnable threw would otherwise leave those
// registrations live: a later successful enable re-registers them, so each event then dispatches
// to the plugin once per failed attempt. Drop them here. Safe on this path only β an
// already-enabled plugin returns early above, so the catch only runs for an enable that never
// went live, which owns no hooks worth keeping. (Idempotent: no-ops when none were registered.)
this.hookManager.unregisterPlugin(pluginId);
throw error;
} finally {
this.enabling.delete(pluginId);
}
}
async disablePlugin(pluginId: string): Promise<void> {
const plugin = this.plugins.get(pluginId);
if (!plugin) {
throw new Error(`Plugin ${pluginId} not found`);
}
if (plugin.status !== PluginStatus.ENABLED) {
return; // Not enabled
}
try {
const host = this.sandboxHosts.get(pluginId);
if (host) {
// Disable is a force-teardown: even if the plugin's onDisable hangs (now bounded) or throws,
// we still kill the worker and drop the reference, so a misbehaving plugin can never block a
// disable or leak its worker thread.
try {
await host.runLifecycle('onDisable', SANDBOX_LIFECYCLE_TIMEOUT_MS);
} catch (error) {
this.logger.warn(`Sandboxed plugin ${pluginId} onDisable failed during disable; terminating anyway`, {
pluginId,
action: 'sandbox_disable_lifecycle_failed',
error: error instanceof Error ? error.message : String(error),
});
}
await host.terminate().catch(() => undefined);
this.sandboxHosts.delete(pluginId);
} else {
const context = this.createPluginContext(plugin);
if (plugin.instance?.onDisable) {
await plugin.instance.onDisable(context);
}
}
// Unregister all hooks for this plugin
this.hookManager.unregisterPlugin(pluginId);
// Drop the plugin's search-provider entry (if any) so queries don't route to a terminated worker.
unregisterPluginSearchProvider(this.getSearchRegistry(), pluginId);
plugin.status = PluginStatus.DISABLED;
this.pluginStorage.setPluginStatus(pluginId, PluginStatus.DISABLED);
this.logger.log(`Plugin disabled: ${plugin.manifest.name}`, {
pluginId,
action: 'plugin_disabled',
});
} catch (error) {
plugin.status = PluginStatus.ERROR;
plugin.error = error instanceof Error ? error.message : String(error);
throw error;
}
}
async unloadPlugin(pluginId: string): Promise<void> {
const plugin = this.plugins.get(pluginId);
if (!plugin) {
throw new Error(`Plugin ${pluginId} not found`);
}
// Disable first if enabled
if (plugin.status === PluginStatus.ENABLED) {
await this.disablePlugin(pluginId);
}
// Call onUnload
if (plugin.instance?.onUnload) {
const context = this.createPluginContext(plugin);
await plugin.instance.onUnload(context);
}
this.plugins.delete(pluginId);
this.logger.log(`Plugin unloaded: ${plugin.manifest.name}`, {
pluginId,
action: 'plugin_unloaded',
});
}
/** Absolute path of the directory user plugins are loaded from (used by install/uninstall). */
getPluginsDir(): string {
return this.pluginsDir;
}
/** Whether a plugin is a first-party built-in (engine / bundled extension) vs an installed user plugin. */
isBuiltIn(pluginId: string): boolean {
return this.pluginStorage.getPluginEntry(pluginId)?.builtIn ?? false;
}
/**
* Fully remove an installed user plugin: disable + unload from the runtime, drop its persisted
* registry entry, and delete its directory from disk. Built-ins (engines, bundled extensions) are
* registered programmatically with no on-disk dir and must never be removable.
*/
async uninstallPlugin(pluginId: string): Promise<void> {
if (this.pluginStorage.getPluginEntry(pluginId)?.builtIn) {
throw new Error(`Cannot uninstall built-in plugin ${pluginId}`);
}
if (this.plugins.has(pluginId)) {
await this.unloadPlugin(pluginId);
}
this.pluginStorage.deletePluginEntry(pluginId);
// Delete the plugin's directory, guarding against a traversal id escaping the plugins dir.
const base = path.resolve(this.pluginsDir);
const dir = path.resolve(base, pluginId);
if (dir !== base && dir.startsWith(base + path.sep) && fs.existsSync(dir)) {
fs.rmSync(dir, { recursive: true, force: true });
}
this.logger.log(`Plugin uninstalled: ${pluginId}`, { pluginId, action: 'plugin_uninstalled' });
}
updatePluginConfig(pluginId: string, config: Record<string, unknown>): void {
const plugin = this.plugins.get(pluginId);
if (!plugin) {
throw new Error(`Plugin ${pluginId} not found`);
}
plugin.config = { ...plugin.config, ...config };
// Persist config
this.pluginStorage.setPluginConfig(pluginId, plugin.config);
// Notify the running plugin of the config change (fire and forget). A sandboxed plugin's
// onConfigChange lives in the worker (plugin.instance is null), so route it through the live worker
// host so it refreshes ctx.config too; built-ins go through the in-process instance.
if (plugin.status === PluginStatus.ENABLED) {
const sandboxHost = this.sandboxHosts.get(pluginId);
if (sandboxHost) {
sandboxHost.sendConfigChange(plugin.config);
} else if (plugin.instance?.onConfigChange) {
const context = this.createPluginContext(plugin);
void plugin.instance.onConfigChange(context, plugin.config);
}
}
this.logger.debug(`Plugin config updated: ${pluginId}`, {
pluginId,
action: 'plugin_config_updated',
});
}
/**
* Set the sessions a session-scoped plugin is activated for. `['*']` = all numbers (system-wide),
* an explicit list scopes it to those sessions, `[]` deactivates it everywhere. Takes effect on the
* next hook event (the gate reads plugin.activeSessions live) and survives a restart.
*/
setPluginSessions(pluginId: string, sessions: string[]): PluginInstance {
const plugin = this.plugins.get(pluginId);
if (!plugin) {
throw new Error(`Plugin ${pluginId} not found`);
}
if (plugin.manifest.sessionScoped === false) {
throw new Error(`Plugin ${pluginId} is global (not session-scoped) and cannot be activated per session`);
}
plugin.activeSessions = sessions;
this.pluginStorage.setPluginSessions(pluginId, sessions);
this.logger.log(`Plugin active sessions updated: ${pluginId}`, {
pluginId,
action: 'plugin_sessions_updated',
sessions,
});
return plugin;
}
/**
* Set (or clear) a plugin's per-session config override for `sessionId`. Hooks for that session then
* see the override shallow-merged over the base via ctx.config β applied on the next event
* (resolution reads plugin.sessionConfig live) and persisted across restart. An empty override
* removes it (the session falls back to the base). Global plugins have no per-session config.
*/
setPluginSessionConfig(pluginId: string, sessionId: string, config: Record<string, unknown>): PluginInstance {
const plugin = this.plugins.get(pluginId);
if (!plugin) {
throw new Error(`Plugin ${pluginId} not found`);
}
if (plugin.manifest.sessionScoped === false) {
throw new Error(`Plugin ${pluginId} is global (not session-scoped) and has no per-session config`);
}
const next = { ...(plugin.sessionConfig ?? {}) };
if (config && Object.keys(config).length > 0) {
next[sessionId] = config;
} else {
delete next[sessionId];
}
plugin.sessionConfig = next;
this.pluginStorage.setPluginSessionConfig(pluginId, next);
this.logger.debug(`Plugin session config updated: ${pluginId}`, {
pluginId,
action: 'plugin_session_config_updated',
sessionId,
});
return plugin;
}
/**
* Run a plugin's healthCheck across both tiers. A sandboxed plugin's healthCheck lives in the worker
* (plugin.instance is null), so route to the live worker host (time-bounded); built-ins use the
* in-process instance. Returns the default "healthy" when the plugin implements no health check.
*/
async checkPluginHealth(pluginId: string): Promise<{ healthy: boolean; message?: string }> {
const sandboxHost = this.sandboxHosts.get(pluginId);
if (sandboxHost) {
return sandboxHost.healthCheck(SANDBOX_HEALTH_TIMEOUT_MS);
}
const plugin = this.plugins.get(pluginId);
if (plugin?.instance?.healthCheck) {
return plugin.instance.healthCheck();
}
return { healthy: true, message: 'Plugin does not implement health check' };
}
/**
* Dispatch a queued ingress job into its plugin's live sandbox worker. Called from IngressProcessor,
* mirroring checkPluginHealth's sandboxHosts lookup. Throws when the plugin has no live
* worker (disabled/crashed since the job was enqueued) or when the worker's handler itself reports
* failure (`!result.ok`, e.g. a 502/504/500) β either way BullMQ's retry/DLQ machinery takes over.
*/
async dispatchWebhookForInstance(d: IngressJobData): Promise<void> {
const host = this.sandboxHosts.get(d.pluginId);
if (!host) {
throw new Error('no live sandbox host for plugin ' + d.pluginId);
}
// Resolve this instance's per-session config (the base merged with the sessionScope override that
// provisioning wrote) so the ingress handler reads it as ctx.config β this is what makes a minted
// instance multi-tenant. Best-effort: an unresolved plugin just yields undefined (base config only).
const plugin = this.plugins.get(d.pluginId);
const route = plugin?.manifest.ingress?.find(candidate => candidate.route === d.route);
// Reaching dispatch means every authenticating scheme already passed host verification. A route
// explicitly configured with scheme:none is unauthenticated and must never be labelled verified.
// Missing/hot-swapped route metadata fails closed.
const verified = route ? route.signature.scheme !== 'none' : false;
const instance = await this.getPluginInstanceService().resolve(d.pluginId, d.instanceId);
const config = plugin
? resolvePluginConfig(
plugin.config,
plugin.sessionConfig,
instance?.sessionScope ?? undefined,
plugin.manifest.sessionScoped !== false,
)
: undefined;
const result = await host.dispatchWebhook({
instanceId: d.instanceId,
route: d.route,
method: d.method ?? 'POST',
headers: d.payload.headers,
query: d.payload.query,
body: d.payload.body,
rawBody: d.payload.rawBody,
verified,
deliveryId: d.deliveryId,
sessionId: d.sessionId,
config,
timeoutMs: INGRESS_DISPATCH_TIMEOUT_MS,
});
if (!result.ok) {
throw new Error(result.error ?? 'ingress dispatch failed with status ' + result.status);
}
}
/**
* Resolve MessageService at call time via a lazy require so plugin-loader creates NO top-level
* module-load edge to message.service. A static import closes the cycle
* plugin-loader -> message -> session -> engine.factory -> core/plugins barrel -> plugin-loader,
* which corrupts MessageService's constructor paramtype metadata (SessionService -> undefined) at boot.
*/
private getMessageService(): MessageService {
const mod =
// eslint-disable-next-line @typescript-eslint/no-require-imports
require('../../modules/message/message.service') as typeof import('../../modules/message/message.service');
return this.moduleRef.get(mod.MessageService, { strict: false });
}
private getSessionService(): SessionService {
const mod =
// eslint-disable-next-line @typescript-eslint/no-require-imports
require('../../modules/session/session.service') as typeof import('../../modules/session/session.service');
return this.moduleRef.get(mod.SessionService, { strict: false });
}
/**
* Same lazy-require pattern as getMessageService/getSessionService: a static import of the
* integration module would add a top-level edge back into plugin-loader's own module graph.
*/
private getConversationMappingService(): ConversationMappingService {
const mod =
// eslint-disable-next-line @typescript-eslint/no-require-imports
require('../../modules/integration/conversation-mapping.service') as typeof import('../../modules/integration/conversation-mapping.service');
return this.moduleRef.get(mod.ConversationMappingService, { strict: false });
}
private getPluginInstanceService(): PluginInstanceService {
const mod =
// eslint-disable-next-line @typescript-eslint/no-require-imports
require('../../modules/integration/plugin-instance.service') as typeof import('../../modules/integration/plugin-instance.service');
return this.moduleRef.get(mod.PluginInstanceService, { strict: false });
}
/**
* Resolve the SearchProviderRegistry lazily β search is conditionally loaded (SEARCH_ENABLED=false omits
* SearchModule), so the registry may not be registered. Mirrors the lazy-require pattern for
* MessageService/SessionService to avoid a static module edge and a DI cycle. Returns undefined when
* search is disabled, so the loader can no-op search-provider registration without throwing.
*/
private getSearchRegistry(): SearchProviderRegistry | undefined {
try {
const mod =
// eslint-disable-next-line @typescript-eslint/no-require-imports
require('../../modules/search/search-provider.registry') as typeof import('../../modules/search/search-provider.registry');
return this.moduleRef.get(mod.SearchProviderRegistry, { strict: false });
} catch {
return undefined;
}
}
/**
* Enforce a plugin's declared manifest permissions at the capability boundary. A plugin may only
* use a capability whose permission string it declares in `manifest.permissions`; anything else
* (including a manifest with no permissions) is denied. Runs first in each capability verb so a
* missing grant fails fast and uniformly as a PluginCapabilityError.
*/
private assertPermission(manifest: PluginManifest, permission: PluginCapabilityPermission): void {
if (!(manifest.permissions ?? []).includes(permission)) {
throw new PluginCapabilityError(
`Plugin ${manifest.id} is missing the '${permission}' permission required for this capability`,
);
}
}
/**
* Enforce a plugin's manifest session scope. Runs BEFORE any engine/message resolution β
* sessionId is supplied by the plugin, so this is the security boundary. Absent = ['*'].
*/
private assertSessionAllowed(manifest: PluginManifest, sessionId: string): void {
const allowed = manifest.sessions ?? ['*'];
if (!allowed.includes('*') && !allowed.includes(sessionId)) {
throw new PluginCapabilityError(`Plugin ${manifest.id} is not permitted to act on session ${sessionId}`);
}
}
/** Per-session activation gate: is this plugin currently activated for `sessionId`'s event? */
private isHookActive(plugin: PluginInstance, sessionId: string | undefined): boolean {
return isPluginActiveForSession(plugin.manifest.sessionScoped ?? true, plugin.activeSessions ?? ['*'], sessionId);
}
/**
* The capability session gate. A plugin may act on `sessionId` only if BOTH hold: its manifest scope
* allows the session (the static author boundary, assertSessionAllowed) AND the operator has activated
* the plugin for that session (the dynamic boundary, the same gate hook dispatch uses). manifest.sessions
* alone is not enough β a general adapter ships `['*']` and is scoped by operator activation, so without
* the activeSessions check a plugin activated for one session could reach another's engine/mappings/
* handover. Defaults (`activeSessions ?? ['*']`, `sessionScoped:false`) preserve every unrestricted flow.
*/
private assertSessionActive(plugin: PluginInstance, sessionId: string): void {
this.assertSessionAllowed(plugin.manifest, sessionId);
if (!this.isHookActive(plugin, sessionId)) {
throw new PluginCapabilityError(`Plugin ${plugin.manifest.id} is not activated for session ${sessionId}`);
}
}
/**
* Scope-check, then resolve the live engine for a session. getEngine returns undefined for an
* unknown OR unstarted session (no throw), so guard it into a defined PluginCapabilityError.
* A present-but-not-READY engine throws EngineNotReadyError from the adapter on use (β 409).
*/
private resolveEngine(plugin: PluginInstance, sessionId: string): IWhatsAppEngine {
this.assertSessionActive(plugin, sessionId);
const engine = this.getSessionService().getEngine(sessionId);
if (!engine) {
throw new PluginCapabilityError(`Session ${sessionId} has no active engine (unknown or not started)`);
}
return engine;
}
/** Engine read capabilities: require the `engine:read` permission, then resolve the live engine. */
private resolveEngineRead(plugin: PluginInstance, sessionId: string): IWhatsAppEngine {
this.assertPermission(plugin.manifest, PluginCapabilityPermission.ENGINE_READ);
return this.resolveEngine(plugin, sessionId);
}
/**
* Build a worker host for a sandboxed (untrusted) plugin. Overridable so tests can inject a fake
* instead of spawning a real OS thread. Production loads the compiled worker bootstrap from dist.
*/
protected createSandboxHost(
capDispatcher?: (verb: string, args: unknown[]) => Promise<unknown>,
onHookSubscribe?: (event: string, priority?: number) => void,
onWebhookSubscribe?: (route: string) => void,
onLog?: (level: PluginLogLevel, message: string, meta?: Record<string, unknown>) => void,
runWithHookGuard?: (inFlightEvents: string[], run: () => Promise<unknown>) => Promise<unknown>,
onSearchProviderRegister?: () => void,
onWorkerExit?: (code: number, intentional: boolean) => void,
): PluginWorkerHost {
const workerEntry = path.join(__dirname, 'sandbox', 'worker-bootstrap.js');
return new PluginWorkerHost(
new WorkerThreadChannel({
workerEntry,
maxOldGenerationSizeMb: SANDBOX_MAX_OLD_GEN_MB,
// Withhold host secrets: the worker gets a minimal allowlisted env, not a copy of process.env.
env: buildSandboxWorkerEnv(),
}),
capDispatcher,
onHookSubscribe,
onWebhookSubscribe,
onLog,
runWithHookGuard,
SANDBOX_MAX_INFLIGHT_CAPS,
onSearchProviderRegister,
onWorkerExit,
);
}
/** Built-in (trusted) enable: require + run the lifecycle in-process with the live capability context. */
private async enableInProcess(pluginId: string, plugin: PluginInstance): Promise<void> {
const context = this.createPluginContext(plugin);
if (!plugin.instance) {
// Containment guard: reject a manifest.main that escapes the plugin dir.
const mainPath = resolvePluginMainPath(this.pluginsDir, pluginId, plugin.manifest.main);
// eslint-disable-next-line @typescript-eslint/no-require-imports
const pluginModule = require(mainPath) as { default?: new () => IPlugin };
if (pluginModule.default) {
plugin.instance = new pluginModule.default();
} else {
throw new Error(`Plugin ${pluginId} does not export a default class`);
}
}
if (plugin.instance.onLoad) {
await plugin.instance.onLoad(context);
}
if (plugin.instance.onEnable) {
await plugin.instance.onEnable(context);
}
}
/**
* Untrusted enable: load the plugin in an isolated worker and drive its lifecycle there. Capability
* calls and hooks round-trip to the host, which enforces permission + session scope. A failure
* tears the worker back down.
*/
private async enableSandboxed(pluginId: string, plugin: PluginInstance): Promise<void> {
// Containment guard: reject a manifest.main that escapes the plugin dir.
const mainPath = resolvePluginMainPath(this.pluginsDir, pluginId, plugin.manifest.main);
// The capability dispatcher runs a worker request through the SAME context an in-process plugin
// gets, so permission + session-scope checks (assertPermission / assertSessionActive) apply
// identically. The worker can only ask; the host is the gatekeeper.
const context = this.createPluginContext(plugin);
// When the worker subscribes to a hook, register a shim with the hook manager that dispatches the
// event into the worker (time-bounded, so a wedged plugin can't stall the chain). The shim looks
// the host up at fire time, so disabling the plugin (which removes it + unregisters hooks) stops it.
// Harden the IPC boundary against an untrusted worker flooding the host hook registry. HookEvent is
// a type-only union and the wire payload is an arbitrary string, so a hostile/buggy worker can post
// 'hook-subscribe' with (a) the same event repeatedly and (b) unbounded fabricated event names
// ('x:0','x:1',β¦). Without guards each call adds a live host-side registration (unbounded host-heap
// growth + an O(n log n) re-sort). Three guards, all local to this enableSandboxed call (dropped on
// disable): reject unknown events (bounds growth to the finite known set + drops events that can
// never fire), dedup per event, and a belt-and-suspenders size cap.
const subscribedEvents = new Set<HookEvent>();
let unknownEventWarned = false;
const onHookSubscribe = (event: string, priority?: number): void => {
if (!isKnownHookEvent(event)) {
if (!unknownEventWarned) {
unknownEventWarned = true; // warn at most once per plugin so a flood isn't a log-flood vector
this.logger.warn(`Sandboxed plugin ${pluginId} subscribed to an unknown hook event; ignoring`, {
pluginId,
event,
action: 'sandbox_unknown_hook_event',
});
}
return;
}
if (subscribedEvents.has(event)) return;
if (subscribedEvents.size >= KNOWN_HOOK_EVENTS.size) return; // can't exceed the known set
subscribedEvents.add(event);
this.hookManager.register(
pluginId,
event,
async hookCtx => {
const liveHost = this.sandboxHosts.get(pluginId);
if (!liveHost) return { continue: true };
// Per-session activation gate: a session-scoped plugin only sees events for the sessions
// it is activated for. Pass-through (don't dispatch into the worker) otherwise.
if (!this.isHookActive(plugin, hookCtx.sessionId)) return { continue: true };
// Handover gate: once a human has taken over (or closed) a conversation, the bot stops
// seeing its inbound messages. Scoped to message:received only β every other hook event is
// unaffected. Best-effort + fail-open: a lookup failure (or an event/mapping shape the gate
// can't resolve) must never block a normal message from reaching the adapter.
if (event === 'message:received') {
try {
const chatId = (hookCtx.data as { chatId?: string } | undefined)?.chatId;
if (chatId && hookCtx.sessionId) {
const handover = await this.getConversationMappingService().findHandoverForChat(
hookCtx.sessionId,
chatId,
);
if (!shouldDispatchToPlugin(handover, pluginId)) return { continue: true };
}
} catch (error) {
this.logger.debug(`Handover gate lookup failed for plugin ${pluginId}; dispatching normally`, {
pluginId,
event,
error: error instanceof Error ? error.message : String(error),
action: 'handover_gate_fail_open',
});
}
}
return liveHost
.dispatchHook({
event,
data: hookCtx.data,
sessionId: hookCtx.sessionId,
source: hookCtx.source,
// The host resolves the per-session slice (real secrets β the worker is the plugin's
// trusted execution context) and ships it; the worker exposes it as ctx.config.
config: resolvePluginConfig(
plugin.config,
plugin.sessionConfig,
hookCtx.sessionId,
plugin.manifest.sessionScoped !== false,
),
timeoutMs: SANDBOX_HOOK_TIMEOUT_MS,
onTimeout: () =>
this.logger.warn(`Sandboxed plugin ${pluginId} hook '${event}' timed out`, {
pluginId,
event,
action: 'sandbox_hook_timeout',
}),
})
.then(result => ({ continue: result.continue, data: result.data }));
},
priority,
);
};
// When the worker claims an ingress route, record it against the manifest-declared routes so the
// host knows which routes this worker will handle. Same hardening as onHookSubscribe (the wire
// `route` is an arbitrary untrusted string): drop when the manifest lacks 'webhook:ingress', drop
// an undeclared route (warn once), dedup, and cap. subscribedRoutes is local to this enable call,
// so it is dropped on disable exactly as subscribedEvents is.
const subscribedRoutes = new Set<string>();
const declaredRoutes = new Set((plugin.manifest.ingress ?? []).map(r => r.route));
const onWebhookSubscribe = makeOnWebhookSubscribe({
pluginId,
declaredRoutes,
hasPermission: (plugin.manifest.permissions ?? []).includes(PluginCapabilityPermission.WEBHOOK_INGRESS),
subscribed: subscribedRoutes,
maxRoutes: declaredRoutes.size,
warn: (message, meta) => this.logger.warn(message, meta),
});
// Route the worker plugin's ctx.logger.* calls to the same per-plugin logger an in-process plugin
// uses, so sandboxed plugins log identically (prefixed + structured) instead of bare stdout.
const onLog = (level: PluginLogLevel, message: string, meta?: Record<string, unknown>): void => {
if (level === 'error') context.logger.error(message, undefined, meta);
else context.logger[level](message, meta);
};
// When the worker declares itself a search provider (ctx.registerSearchProvider β
// search-provider-register), register a PluginSearchProvider in the SearchProviderRegistry. The host
// is in sandboxHosts by the time registration fires (during onLoad/onEnable), so look it up lazily
// like onHookSubscribe. Search disabled (no registry, or SEARCH_PROVIDER=none) β the util skips.
const onSearchProviderRegister = (): void => {
const liveHost = this.sandboxHosts.get(pluginId);
if (!liveHost) return;
registerPluginSearchProvider({
pluginId,
label: `${plugin.manifest.name} (plugin)`,
transport: liveHost,
timeoutMs: SANDBOX_SEARCH_TIMEOUT_MS,
registry: this.getSearchRegistry(),
mode: this.configService.get<string>('search.provider', 'auto'),
});
};
// A worker that crashes AFTER a successful enable is otherwise invisible to the loader (handleExit only
// drains in-flight calls). Drop the plugin's search-provider entry so the registry falls back to
// builtin-fts instead of routing every /search to a dead worker (auto mode would otherwise pin the dead
// provider ACTIVE). Mirrors the enable-failure cleanup. Broader crash-lifecycle cleanup (status, hooks)
// is a pre-existing gap for all bridges and out of scope here.
const onWorkerExit = (code: number, intentional: boolean): void => {
// Always release the search-provider slot so the registry can fall back to builtin-fts. On a crash
// this is the only cleanup; on a deliberate disable/enable-failure the explicit unregister already
// ran, making this a harmless no-op.
unregisterPluginSearchProvider(this.getSearchRegistry(), pluginId);
if (intentional) return; // routine disable/enable-failure already logged and expected
// Unexpected crash after a successful enable: the worker is gone. Drop the dead host +
// unregister the hook shims (so they don't keep dispatching into the dead worker) + mark the
// plugin ERROR so the dashboard reflects reality. The dispatchHook/dispatchWebhook dead-checks
// fail-fast; this cleanup is the root-cause fix (it also makes the shim's !liveHost guard fire).
const crashed = this.plugins.get(pluginId);
if (crashed) {
crashed.status = PluginStatus.ERROR;
crashed.error = `worker exited unexpectedly (code ${code})`;
this.pluginStorage.setPluginStatus(pluginId, PluginStatus.ERROR);
}
this.hookManager.unregisterPlugin(pluginId);
this.sandboxHosts.delete(pluginId);
this.logger.warn(`Sandboxed plugin ${pluginId} worker exited unexpectedly (code ${code})`, {
pluginId,
code,
action: 'sandbox_worker_exit',
});
};
const host = this.createSandboxHost(
(verb, args) => dispatchCapabilityVerb(context, verb, args),
onHookSubscribe,
onWebhookSubscribe,
onLog,
// Re-establish the in-flight hook context for worker-initiated capability calls, so a sandboxed
// plugin that sends from within a send hook can't loop the event back into itself unboundedly.
(events, run) => this.hookManager.runInFlight(events as HookEvent[], run),
onSearchProviderRegister,
onWorkerExit,
);
this.sandboxHosts.set(pluginId, host);
try {
await host.load(mainPath, { pluginId, config: plugin.config }, SANDBOX_LIFECYCLE_TIMEOUT_MS);
await host.runLifecycle('onLoad', SANDBOX_LIFECYCLE_TIMEOUT_MS);
await host.runLifecycle('onEnable', SANDBOX_LIFECYCLE_TIMEOUT_MS);
} catch (error) {
this.sandboxHosts.delete(pluginId);
// Drop a search provider registered mid-onEnable before the failure: without this, a plugin that
// registers then throws leaves a dead provider as the ACTIVE registry entry in auto mode, so every
// /search routes to a terminated worker β outage. Mirrors disablePlugin's cleanup.
unregisterPluginSearchProvider(this.getSearchRegistry(), pluginId);
await host.terminate().catch(() => undefined);
throw error;
}
}
private createPluginContext(plugin: PluginInstance): PluginContext {
const pluginLogger: PluginLogger = {
log: (message, meta) =>
this.logger.log(`[${plugin.manifest.id}] ${message}`, { ...meta, pluginId: plugin.manifest.id }),
debug: (message, meta) =>
this.logger.debug(`[${plugin.manifest.id}] ${message}`, { ...meta, pluginId: plugin.manifest.id }),
warn: (message, meta) =>
this.logger.warn(`[${plugin.manifest.id}] ${message}`, { ...meta, pluginId: plugin.manifest.id }),
error: (message, error, meta) =>
this.logger.error(
`[${plugin.manifest.id}] ${message}`,
error instanceof Error ? error.message : String(error),
{ ...meta, pluginId: plugin.manifest.id },
),
};
const hookSession = this.hookSession;
return {
pluginId: plugin.manifest.id,
manifest: plugin.manifest,
// Per-session: inside a hook, returns the override merged over the base for the firing session;
// outside a hook (lifecycle), the base config. A getter so it reflects live config edits too.
get config() {
return resolvePluginConfig(
plugin.config,
plugin.sessionConfig,
hookSession.getStore()?.sessionId,
plugin.manifest.sessionScoped !== false,
);
},
hookManager: this.hookManager,
logger: pluginLogger,
storage: this.pluginStorage.createPluginStorage(plugin.manifest.id),
registerHook: (event, handler, priority) => {
// Wrap with the per-session activation gate so an in-process plugin only handles events for
// the sessions it is activated for (mirrors the sandboxed shim), and scope the firing
// sessionId so ctx.config resolves the right per-session slice for the handler.
this.hookManager.register(
plugin.manifest.id,
event,
async hookCtx => {
if (!this.isHookActive(plugin, hookCtx.sessionId)) return { continue: true };
return this.hookSession.run({ sessionId: hookCtx.sessionId }, () => handler(hookCtx));
},
priority,
);
},
// In-process built-ins are not reached by the ingress pipeline (it dispatches to sandbox hosts),
// so fail loud rather than silently never firing. Sandboxed plugins get a real registerWebhook
// from the worker bootstrap.
registerWebhook: () => {
throw new PluginCapabilityError(
`Plugin ${plugin.manifest.id}: registerWebhook (ingress) is only available to sandboxed plugins`,
);
},
messages: {
sendText: async (sessionId, chatId, text) => {
// Validate permission + scope + that the session has a live engine BEFORE MessageService
// persists a pending row: a missing grant / dead session must fail with
// PluginCapabilityError, not a raw TypeError + orphaned row. resolveEngine also runs
// assertSessionActive.
this.assertPermission(plugin.manifest, PluginCapabilityPermission.MESSAGES_SEND);
this.resolveEngine(plugin, sessionId);
return this.getMessageService().sendText(sessionId, { chatId, text });
},
reply: async (sessionId, chatId, quotedMessageId, text) => {
this.assertPermission(plugin.manifest, PluginCapabilityPermission.MESSAGES_SEND);
this.resolveEngine(plugin, sessionId);
return this.getMessageService().reply(sessionId, { chatId, quotedMessageId, text });
},
} satisfies PluginMessagingCapability,
engine: {
getGroupInfo: async (sessionId, groupId) => this.resolveEngineRead(plugin, sessionId).getGroupInfo(groupId),
getContacts: async sessionId => this.resolveEngineRead(plugin, sessionId).getContacts(),
getContactById: async (sessionId, contactId) =>
this.resolveEngineRead(plugin, sessionId).getContactById(contactId),
checkNumberExists: async (sessionId, phone) =>
this.resolveEngineRead(plugin, sessionId).checkNumberExists(phone),
getChats: async sessionId => this.resolveEngineRead(plugin, sessionId).getChats(),
getChatHistory: async (sessionId, chatId, limit, includeMedia) =>
this.resolveEngineRead(plugin, sessionId).getChatHistory(
chatId,
// Clamp to the REST non-deep ceiling (MessageService.MAX_CHAT_HISTORY_LIMIT = 100) so an
// untrusted plugin can't request an unbounded history fetch.
Math.min(Math.max(Math.trunc(limit ?? 50), 1), 100),
includeMedia ?? false,
),
canonicalChatId: (sessionId, chatId) => {
// resolveEngineRead is the gate only (engine:read permission + live session); the resolution
// itself is a synchronous host lid->phone lookup, not an engine call, mirroring the webhook
// from-filter. Not `async` (nothing to await) β a resolved promise satisfies the signature.
this.resolveEngineRead(plugin, sessionId);
return Promise.resolve(toNeutralJid(chatId, jid => this.lidMappingStore?.getCached(userPart(jid)) ?? null));
},
} satisfies PluginEngineReadCapability,
net: {
fetch: async (url, init) => {
// Two gates: the declared permission, then the effective host allowlist = manifest net.allow
// UNION the hosts of net.allowConfigHosts keys across the base config AND every per-session
// override. The host gate has no firing-session context for a sandboxed plugin's cap round-trip,
// so admit every operator-configured tenant host (all public + still SSRF-guarded at connect)
// rather than resolving a single, possibly wrong (base-only), one. The SSRF guard inside
// performPluginFetch still blocks internal IPs even when the host is allowlisted.
this.assertPermission(plugin.manifest, PluginCapabilityPermission.NET_FETCH);
const netConfigs = [plugin.config ?? {}, ...Object.values(plugin.sessionConfig ?? {})];
const allow = [
...new Set(
netConfigs.flatMap(cfg =>
effectiveNetAllow(plugin.manifest.net?.allow, plugin.manifest.net?.allowConfigHosts, cfg),
),
),
];
if (!isNetHostAllowed(allow, url)) {
throw new PluginCapabilityError(
`Plugin ${plugin.manifest.id} may not fetch ${url} β add its host to net.allow or net.allowConfigHosts`,
);
}
return performPluginFetch(url, init);
},
} satisfies PluginNetCapability,
conversations: buildConversationSendFacade({
manifest: plugin.manifest,
assertPermission: this.assertPermission.bind(this),
assertSessionActive: (sessionId: string) => this.assertSessionActive(plugin, sessionId),
resolveChatId: async env => {
if (!env.instanceId || !env.source?.externalConversationId) {
throw new PluginCapabilityError(
`Plugin ${plugin.manifest.id}: conversation.send requires chatId, or both instanceId and source to resolve one`,
);
}
const mapping = await this.getConversationMappingService().getByProvider(
plugin.manifest.id,
env.instanceId,
env.source.externalConversationId,
);
if (!mapping) {
throw new PluginCapabilityError(
`Plugin ${plugin.manifest.id}: no conversation mapping for instance ${env.instanceId} / ${env.source.externalConversationId}`,
);
}
return mapping.chatId;
},
// Re-establish the in-flight hook context around the downstream send so an adapter that calls
// conversation.send from within its own ingress handling can't echo-loop back into itself via
// its own outbound message:sending hook. Gate on an ALREADY-in-flight event (mirrors the
// worker-cap wrap's `inFlight.length > 0` check): a plain top-level send must NOT suppress
// message:sending for unrelated observers (audit/moderation) β only genuine re-entrancy does.
runGuarded: (events, run) =>
(events as HookEvent[]).some(e => this.hookManager.isInFlight(e))
? this.hookManager.runInFlight(events as HookEvent[], run)
: run(),
sendText: (sessionId, opts) => this.getMessageService().sendText(sessionId, opts),
reply: (sessionId, opts) => this.getMessageService().reply(sessionId, opts),
sendMedia: (sessionId, opts) => dispatchConversationMedia(this.getMessageService(), sessionId, opts),
} satisfies Parameters<typeof buildConversationSendFacade>[0]) satisfies PluginConversationsCapability,
handover: {
set: async (key, state) => {
// Same gate as conversation.send: flipping handover is part of owning the conversation, so
// it reuses CONVERSATION_SEND rather than adding a new permission.
this.assertPermission(plugin.manifest, PluginCapabilityPermission.CONVERSATION_SEND);
this.assertSessionActive(plugin, key.sessionId);
const mapping = await this.getConversationMappingService().get({
sessionId: key.sessionId,
chatId: key.chatId,
pluginId: plugin.manifest.id,
instanceId: key.instanceId,
});
if (!mapping) {
throw new PluginCapabilityError(
`Plugin ${plugin.manifest.id}: no conversation mapping for session ${key.sessionId} / chat ${key.chatId} / instance ${key.instanceId}`,
);
}
await this.getConversationMappingService().setHandover(mapping.id, state);
},
} satisfies PluginHandoverCapability,
mappings: {
upsert: async (key, providerConversationId) => {
this.assertPermission(plugin.manifest, PluginCapabilityPermission.CONVERSATION_SEND);
this.assertSessionActive(plugin, key.sessionId);
await this.getConversationMappingService().upsert(
{ sessionId: key.sessionId, chatId: key.chatId, pluginId: plugin.manifest.id, instanceId: key.instanceId },
providerConversationId,
);
},
get: async key => {
this.assertPermission(plugin.manifest, PluginCapabilityPermission.CONVERSATION_SEND);
this.assertSessionActive(plugin, key.sessionId);
const m = await this.getConversationMappingService().get({
sessionId: key.sessionId,
chatId: key.chatId,
pluginId: plugin.manifest.id,
instanceId: key.instanceId,
});
return m ? { providerConversationId: m.providerConversationId, handoverState: m.handoverState } : null;
},
getByProvider: async (instanceId, providerConversationId) => {
this.assertPermission(plugin.manifest, PluginCapabilityPermission.CONVERSATION_SEND);
const m = await this.getConversationMappingService().getByProvider(
plugin.manifest.id,
instanceId,
providerConversationId,
);
// Parity with get/upsert: a plugin may only read a mapping for a session it is activated for.
if (m) this.assertSessionActive(plugin, m.sessionId);
return m ? { sessionId: m.sessionId, chatId: m.chatId, handoverState: m.handoverState } : null;
},
} satisfies PluginMappingsCapability,
};
}
// ============================================================================
// Query Methods
// ============================================================================
getPlugin(pluginId: string): PluginInstance | undefined {
return this.plugins.get(pluginId);
}
getAllPlugins(): PluginInstance[] {
return Array.from(this.plugins.values());
}
getPluginsByType(type: PluginType): PluginInstance[] {
return this.getAllPlugins().filter(p => p.manifest.type === type);
}
getEnabledPlugins(): PluginInstance[] {
return this.getAllPlugins().filter(p => p.status === PluginStatus.ENABLED);
}
isPluginEnabled(pluginId: string): boolean {
const plugin = this.plugins.get(pluginId);
return plugin?.status === PluginStatus.ENABLED;
}
// ============================================================================
// Built-in Plugin Registration (for Phase 4)
// ============================================================================
registerBuiltInPlugin(manifest: PluginManifest, instance: IPlugin, config: Record<string, unknown> = {}): void {
// Merge: env-derived defaults stay live each boot (so a changed .env wins), while an operator's
// persisted overrides win for the keys they actually set. Engine config is wholly env-derived
// (no persisted overrides), so it is never frozen to a first-boot snapshot.
const effectiveConfig = { ...config, ...(this.pluginStorage.getPluginConfig(manifest.id) ?? {}) };
const pluginInstance: PluginInstance = {
manifest,
status: PluginStatus.INSTALLED,
config: effectiveConfig,
instance,
loadedAt: new Date(),
builtIn: true,
// Read persisted per-session activation + config back into the runtime, like loadPlugin β
// otherwise the delivery gate falls back to all-sessions/base-config after every restart for a
// session-scoped built-in the operator had restricted.
activeSessions: this.pluginStorage.getPluginSessions(manifest.id) ?? undefined,
sessionConfig: this.pluginStorage.getPluginSessionConfig(manifest.id) ?? undefined,
};
this.plugins.set(manifest.id, pluginInstance);
// Ensure a registry entry exists so later enable/disable/config writes persist.
this.ensureRegistryEntry(manifest, true);
this.logger.debug(`Built-in plugin registered: ${manifest.name}`, {
pluginId: manifest.id,
action: 'builtin_plugin_registered',
});
}
}
|