feat(plugin): remote runtime

A plugin can now run as a service the administrator deploys elsewhere and
registers by URL (runtime.type: remote). Installing asks for the URL,
issues a 32-byte signing secret (stored sealed, shown once) and keeps both
across upgrades. The admin API and the plugin detail drawer change the URL
and rotate the secret.

- internal/plugin/remote: an Activator that checks the service's health
  and manifest against the package, signs every call with the secret,
  health-checks it and reports outages to the reconciler; after an outage
  it checks the manifest again.
- Calls route to the host or remote runtime by where the plugin loaded.
  Remote plugins get Host API access only when WEKNORA_PLUGIN_HOST_API_URL
  names an address they can reach.
- The reconciler reloads a plugin when its URL or secret changes, and now
  retries a failed activation with backoff (30s doubling to 10m) instead of
  leaving it failed until someone changes it. That also covers host
  plugins that failed to start. A plugin that failed is unloaded when
  disabled.
- SavePlugin updates the new columns; a test keeps the update column list
  in step with the table.
- Migrations: plugins.remote_url, plugins.remote_secret.
This commit is contained in:
wizardchen
2026-09-26 00:23:45 +08:00
parent fa261f3b0a
commit 39ffd07597
33 changed files with 1325 additions and 71 deletions
+28 -7
View File
@@ -51,15 +51,20 @@ export interface InstalledPlugin {
manifest?: PluginManifest
versions: PluginVersion[]
node?: PluginNodeStatus
/** Where a remote plugin's service runs. */
remote_url?: string
/** A remote plugin's new signing secret, only in the response that issued it. */
issuedSecret?: string
}
/** A package to inspect or install: an uploaded file or a URL. */
export type PackageSource = { file: File } | { url: string }
function packageForm(file: File, digest?: string) {
function packageForm(file: File, digest?: string, remoteUrl?: string) {
const form = new FormData()
form.append('file', file)
if (digest) form.append('digest', digest)
if (remoteUrl) form.append('remote_url', remoteUrl)
return form
}
@@ -81,14 +86,21 @@ export function inspectPluginPackage(source: PackageSource) {
return post<{ data: PluginPreview }>(`${BASE}/inspect`, { url: source.url }, { timeout: PACKAGE_TIMEOUT })
}
/** Installs the package reviewed with inspect; digest pins it to that package. */
export function installPluginPackage(source: PackageSource, digest: string) {
/**
* Installs the package reviewed with inspect; digest pins it to that package.
* A remote plugin also needs the URL of its service (optional on upgrades).
*/
export function installPluginPackage(source: PackageSource, digest: string, remoteUrl?: string) {
if ('file' in source) {
return postUpload(BASE, packageForm(source.file, digest), undefined, { timeout: PACKAGE_TIMEOUT }) as Promise<{
data: InstalledPlugin
}>
return postUpload(BASE, packageForm(source.file, digest, remoteUrl), undefined, {
timeout: PACKAGE_TIMEOUT,
}) as Promise<{ data: InstalledPlugin }>
}
return post<{ data: InstalledPlugin }>(BASE, { url: source.url, digest }, { timeout: PACKAGE_TIMEOUT })
return post<{ data: InstalledPlugin }>(
BASE,
{ url: source.url, digest, remote_url: remoteUrl || undefined },
{ timeout: PACKAGE_TIMEOUT },
)
}
export function setInstalledPluginEnabled(id: string, enabled: boolean) {
@@ -99,6 +111,15 @@ export function activatePluginVersion(id: string, version: string) {
return put<{ data: InstalledPlugin }>(`${BASE}/${encodeURIComponent(id)}/active-version`, { version })
}
export function setPluginRemoteUrl(id: string, url: string) {
return put<{ data: InstalledPlugin }>(`${BASE}/${encodeURIComponent(id)}/remote-url`, { url })
}
/** Issues a new signing secret; the response carries it in issuedSecret. */
export function rotatePluginSecret(id: string) {
return post<{ data: InstalledPlugin }>(`${BASE}/${encodeURIComponent(id)}/secret/rotate`, {})
}
export function uninstallPlugin(id: string) {
return del(`${BASE}/${encodeURIComponent(id)}`)
}
+22 -2
View File
@@ -2736,7 +2736,8 @@ export default {
pluginAdmin: {
runtime: {
declarative: 'Declarative',
host: 'Local process'
host: 'Local process',
remote: 'Remote service'
},
title: 'Plugin management',
description: 'Install and manage plugins for the whole platform. Every workspace sees an installed plugin, and each one enables it for itself; disabling a plugin here unloads it on every node.',
@@ -2780,6 +2781,13 @@ export default {
hostApi: 'WeKnora API',
events: 'Events'
},
secret: {
title: 'Plugin signing secret',
description: 'Set this secret as the plugin service\'s WEKNORA_PLUGIN_SECRET environment variable; the service uses it to verify that requests come from WeKnora. It is shown only once and cannot be viewed again after you close this.',
copy: 'Copy',
copied: 'Copied',
done: 'I have saved it'
},
install: {
title: 'Install plugin',
description: 'Upload a .wkp package or give its download URL, review it, then install.',
@@ -2794,6 +2802,9 @@ export default {
fileHint: 'A .wkp file (a zip with plugin.yaml), up to 64 MB.',
urlLabel: 'Download URL',
urlHint: 'The server downloads this URL; private network addresses are refused.',
remoteUrlLabel: 'Service URL',
remoteUrlHint: 'The plugin service\'s HTTP(S) address. Private network hosts must be listed in SSRF_WHITELIST.',
remoteUrlKeep: 'On an upgrade, leave empty to keep the current URL',
reviewSection: 'Review before installing',
noPermissions: 'This plugin asks for no extra permissions.',
configNotice: 'This plugin needs configuration: platform settings are in its details after installing, and workspace admins fill in workspace settings in Plugins.',
@@ -2836,7 +2847,16 @@ export default {
uninstall: 'Uninstall plugin',
uninstallConfirm: 'Uninstall? Every node unloads the plugin right away.',
uninstalled: 'Plugin uninstalled',
uninstallFailed: 'Uninstall failed'
uninstallFailed: 'Uninstall failed',
remote: 'Remote service',
remoteUrl: 'Service URL',
editUrl: 'Change',
urlSaved: 'Service URL updated',
urlSaveFailed: 'Failed to update the service URL',
rotateSecret: 'Rotate secret',
rotateHint: 'Issues a new signing secret. Calls to the service fail until it is given the new one.',
rotateConfirm: 'Rotate the secret? The old one stops working immediately.',
rotateFailed: 'Failed to rotate the secret'
}
},
pluginCenter: {
+22 -2
View File
@@ -2736,7 +2736,8 @@ export default {
pluginAdmin: {
runtime: {
declarative: '宣言型',
host: 'ローカルプロセス'
host: 'ローカルプロセス',
remote: 'リモートサービス'
},
title: 'プラグイン管理',
description: 'プラットフォーム全体のプラグインをインストール・管理します。インストールしたプラグインはすべてのワークスペースに表示され、各ワークスペースで個別に有効化します。ここで無効にすると全ノードでアンロードされます。',
@@ -2780,6 +2781,13 @@ export default {
hostApi: 'WeKnora API',
events: 'イベント'
},
secret: {
title: 'プラグイン署名シークレット',
description: 'このシークレットをプラグインサービスの環境変数 WEKNORA_PLUGIN_SECRET に設定してください。サービスはこれでリクエストが WeKnora からのものか確認します。表示されるのは今回だけで、閉じると再表示できません。',
copy: 'コピー',
copied: 'コピーしました',
done: '保存しました'
},
install: {
title: 'プラグインをインストール',
description: '.wkp パッケージをアップロードするかダウンロード URL を指定し、確認してからインストールします。',
@@ -2794,6 +2802,9 @@ export default {
fileHint: '.wkp ファイル(plugin.yaml を含む zip)、最大 64 MB。',
urlLabel: 'ダウンロード URL',
urlHint: 'サーバーがこの URL からダウンロードします。プライベートネットワークのアドレスは拒否されます。',
remoteUrlLabel: 'サービス URL',
remoteUrlHint: 'プラグインサービスの HTTP(S) アドレス。プライベートネットワークのホストは SSRF_WHITELIST に追加する必要があります。',
remoteUrlKeep: 'アップグレード時は空欄のままで現在の URL を引き続き使用します',
reviewSection: 'インストール前の確認',
noPermissions: 'このプラグインは追加の権限を要求しません。',
configNotice: 'このプラグインには設定が必要です。プラットフォーム設定はインストール後の詳細で、ワークスペース設定は各ワークスペース管理者がプラグインセンターで入力します。',
@@ -2836,7 +2847,16 @@ export default {
uninstall: 'プラグインをアンインストール',
uninstallConfirm: 'アンインストールしますか?すべてのノードで直ちにアンロードされます。',
uninstalled: 'プラグインをアンインストールしました',
uninstallFailed: 'アンインストールに失敗しました'
uninstallFailed: 'アンインストールに失敗しました',
remote: 'リモートサービス',
remoteUrl: 'サービス URL',
editUrl: '変更',
urlSaved: 'サービス URL を更新しました',
urlSaveFailed: 'サービス URL の更新に失敗しました',
rotateSecret: 'シークレットをローテーション',
rotateHint: '新しい署名シークレットを発行します。サービスに新しいシークレットを設定するまで呼び出しは失敗します。',
rotateConfirm: 'ローテーションしますか?古いシークレットは直ちに無効になります。',
rotateFailed: 'シークレットのローテーションに失敗しました'
}
},
pluginCenter: {
+22 -2
View File
@@ -5254,7 +5254,8 @@ export default {
pluginAdmin: {
runtime: {
declarative: '선언형',
host: '로컬 프로세스'
host: '로컬 프로세스',
remote: '원격 서비스'
},
title: '플러그인 관리',
description: '플랫폼 전체의 플러그인을 설치하고 관리합니다. 설치된 플러그인은 모든 워크스페이스에 보이며 각 워크스페이스가 직접 활성화합니다. 여기서 비활성화하면 모든 노드에서 언로드됩니다.',
@@ -5298,6 +5299,13 @@ export default {
hostApi: 'WeKnora API',
events: '이벤트'
},
secret: {
title: '플러그인 서명 비밀 키',
description: '이 비밀 키를 플러그인 서비스의 환경 변수 WEKNORA_PLUGIN_SECRET으로 설정하세요. 서비스는 이를 사용해 요청이 WeKnora에서 온 것인지 확인합니다. 이번에만 표시되며 닫은 후에는 다시 볼 수 없습니다.',
copy: '복사',
copied: '복사했습니다',
done: '저장했습니다'
},
install: {
title: '플러그인 설치',
description: '.wkp 패키지를 업로드하거나 다운로드 URL을 입력하고, 검토한 뒤 설치합니다.',
@@ -5312,6 +5320,9 @@ export default {
fileHint: '.wkp 파일(plugin.yaml이 포함된 zip), 최대 64 MB.',
urlLabel: '다운로드 URL',
urlHint: '서버가 이 URL에서 다운로드합니다. 사설 네트워크 주소는 거부됩니다.',
remoteUrlLabel: '서비스 URL',
remoteUrlHint: '플러그인 서비스의 HTTP(S) 주소입니다. 사설 네트워크 호스트는 SSRF_WHITELIST에 추가해야 합니다.',
remoteUrlKeep: '업그레이드 시 비워 두면 현재 URL을 계속 사용합니다',
reviewSection: '설치 전 검토',
noPermissions: '이 플러그인은 추가 권한을 요청하지 않습니다.',
configNotice: '이 플러그인은 설정이 필요합니다. 플랫폼 설정은 설치 후 상세 화면에서, 워크스페이스 설정은 각 워크스페이스 관리자가 플러그인 센터에서 입력합니다.',
@@ -5354,7 +5365,16 @@ export default {
uninstall: '플러그인 제거',
uninstallConfirm: '제거할까요? 모든 노드에서 즉시 언로드됩니다.',
uninstalled: '플러그인을 제거했습니다',
uninstallFailed: '제거하지 못했습니다'
uninstallFailed: '제거하지 못했습니다',
remote: '원격 서비스',
remoteUrl: '서비스 URL',
editUrl: '변경',
urlSaved: '서비스 URL을 업데이트했습니다',
urlSaveFailed: '서비스 URL을 업데이트하지 못했습니다',
rotateSecret: '비밀 키 교체',
rotateHint: '새 서명 비밀 키를 발급합니다. 서비스에 새 키를 설정하기 전까지 호출이 실패합니다.',
rotateConfirm: '비밀 키를 교체할까요? 이전 키는 즉시 무효화됩니다.',
rotateFailed: '비밀 키를 교체하지 못했습니다'
}
},
pluginCenter: {
+22 -2
View File
@@ -5254,7 +5254,8 @@ export default {
pluginAdmin: {
runtime: {
declarative: 'Декларативный',
host: 'Локальный процесс'
host: 'Локальный процесс',
remote: 'Удалённый сервис'
},
title: 'Управление плагинами',
description: 'Установка и управление плагинами всей платформы. Установленный плагин виден всем рабочим пространствам, и каждое включает его самостоятельно; отключение здесь выгружает плагин на всех узлах.',
@@ -5298,6 +5299,13 @@ export default {
hostApi: 'API WeKnora',
events: 'События'
},
secret: {
title: 'Секрет подписи плагина',
description: 'Укажите этот секрет в переменной окружения WEKNORA_PLUGIN_SECRET сервиса плагина: по нему сервис проверяет, что запросы приходят от WeKnora. Секрет показывается только один раз и после закрытия недоступен.',
copy: 'Копировать',
copied: 'Скопировано',
done: 'Я сохранил его'
},
install: {
title: 'Установка плагина',
description: 'Загрузите пакет .wkp или укажите URL для скачивания, проверьте и установите.',
@@ -5312,6 +5320,9 @@ export default {
fileHint: 'Файл .wkp (zip с plugin.yaml), до 64 МБ.',
urlLabel: 'URL для скачивания',
urlHint: 'Сервер скачает этот URL; адреса частных сетей запрещены.',
remoteUrlLabel: 'URL сервиса',
remoteUrlHint: 'HTTP(S)-адрес сервиса плагина. Хосты частной сети должны быть указаны в SSRF_WHITELIST.',
remoteUrlKeep: 'При обновлении оставьте пустым, чтобы сохранить текущий URL',
reviewSection: 'Проверка перед установкой',
noPermissions: 'Плагин не запрашивает дополнительных разрешений.',
configNotice: 'Плагину нужны настройки: настройки платформы — в его карточке после установки, настройки рабочего пространства заполняют его администраторы в разделе «Плагины».',
@@ -5354,7 +5365,16 @@ export default {
uninstall: 'Удалить плагин',
uninstallConfirm: 'Удалить? Все узлы сразу выгрузят плагин.',
uninstalled: 'Плагин удалён',
uninstallFailed: 'Не удалось удалить'
uninstallFailed: 'Не удалось удалить',
remote: 'Удалённый сервис',
remoteUrl: 'URL сервиса',
editUrl: 'Изменить',
urlSaved: 'URL сервиса обновлён',
urlSaveFailed: 'Не удалось обновить URL сервиса',
rotateSecret: 'Сменить секрет',
rotateHint: 'Выпускает новый секрет подписи. Вызовы сервиса будут завершаться ошибкой, пока он не получит новый секрет.',
rotateConfirm: 'Сменить секрет? Старый сразу перестанет действовать.',
rotateFailed: 'Не удалось сменить секрет'
}
},
pluginCenter: {
+22 -2
View File
@@ -5256,7 +5256,8 @@ export default {
pluginAdmin: {
runtime: {
declarative: '声明式',
host: '本机进程'
host: '本机进程',
remote: '远程服务'
},
title: '插件管理',
description: '安装与管理本平台的插件。安装后所有空间都能看到,但每个空间需自行启用;停用会在所有节点卸载该插件。',
@@ -5300,6 +5301,13 @@ export default {
hostApi: '调用 WeKnora API',
events: '订阅事件'
},
secret: {
title: '插件签名密钥',
description: '把此密钥配置为插件服务的环境变量 WEKNORA_PLUGIN_SECRET,服务据此确认请求来自 WeKnora。密钥只显示这一次,关闭后无法再查看。',
copy: '复制',
copied: '已复制',
done: '我已保存'
},
install: {
title: '安装插件',
description: '上传 .wkp 插件包或填写下载地址,审阅后安装。',
@@ -5314,6 +5322,9 @@ export default {
fileHint: '.wkp 文件(包含 plugin.yaml 的 zip),最大 64 MB。',
urlLabel: '下载地址',
urlHint: '服务端会下载该地址;不允许内网地址。',
remoteUrlLabel: '服务地址',
remoteUrlHint: '插件服务的 HTTP(S) 地址。内网地址需要加入 SSRF_WHITELIST。',
remoteUrlKeep: '升级时可留空,沿用当前地址',
reviewSection: '安装前审阅',
noPermissions: '该插件不申请任何额外权限。',
configNotice: '该插件需要配置:平台配置在安装后的详情中填写,空间配置由各空间管理员在插件中心填写。',
@@ -5356,7 +5367,16 @@ export default {
uninstall: '卸载插件',
uninstallConfirm: '确定卸载?所有节点会立即卸载该插件。',
uninstalled: '插件已卸载',
uninstallFailed: '卸载失败'
uninstallFailed: '卸载失败',
remote: '远程服务',
remoteUrl: '服务地址',
editUrl: '修改',
urlSaved: '服务地址已更新',
urlSaveFailed: '更新服务地址失败',
rotateSecret: '轮换密钥',
rotateHint: '生成新的签名密钥。服务换上新密钥之前,对它的调用会失败。',
rotateConfirm: '确定轮换?旧密钥立即失效。',
rotateFailed: '轮换密钥失败'
}
},
pluginCenter: {
+22 -1
View File
@@ -64,6 +64,18 @@
</div>
<PluginInstallDrawer v-model:visible="installOpen" @installed="upsert" />
<t-dialog
v-model:visible="secretOpen"
:header="t('pluginAdmin.secret.title')"
:confirm-btn="{ content: t('pluginAdmin.secret.copy'), theme: 'primary' }"
:cancel-btn="{ content: t('pluginAdmin.secret.done') }"
:close-on-overlay-click="false"
@confirm="copyWithToast(issuedSecret, 'pluginAdmin.secret.copied')"
@closed="issuedSecret = ''"
>
<p>{{ t('pluginAdmin.secret.description') }}</p>
<t-textarea :value="issuedSecret" readonly autosize />
</t-dialog>
<PluginDetailDrawer
v-model:visible="detailOpen"
:plugin="selected"
@@ -79,6 +91,7 @@ import { useI18n } from 'vue-i18n'
import { MessagePlugin } from 'tdesign-vue-next'
import { listInstalledPlugins, setInstalledPluginEnabled, type InstalledPlugin } from '@/api/system/plugins'
import { copyWithToast } from '@/utils/clipboard'
import { localizedText } from '@/utils/localizedText'
import { contributionSummary } from '../settings/pluginCenterState'
@@ -97,6 +110,9 @@ const loading = ref(false)
const pending = ref(new Set<string>())
const installOpen = ref(false)
const detailOpen = ref(false)
// A remote plugin's signing secret, shown once after it is issued.
const secretOpen = ref(false)
const issuedSecret = ref('')
const selectedId = ref('')
const selected = computed(() => plugins.value.find((p) => p.id === selectedId.value) ?? null)
@@ -129,7 +145,12 @@ async function load() {
}
}
function upsert(p: InstalledPlugin) {
function upsert(received: InstalledPlugin) {
const { issuedSecret: secret, ...p } = received
if (secret) {
issuedSecret.value = secret
secretOpen.value = true
}
const i = plugins.value.findIndex((x) => x.id === p.id)
if (i >= 0) plugins.value.splice(i, 1, p)
else plugins.value = [...plugins.value, p].sort((a, b) => a.id.localeCompare(b.id))
@@ -12,6 +12,7 @@ import {
isPackageUrl,
permissionLines,
remoteHosts,
remoteUrlReady,
shortDigest,
sortVersions,
} from './pluginManagementState'
@@ -77,3 +78,14 @@ test('isPackageUrl and hasSystemConfig', () => {
assert.equal(hasSystemConfig(manifest), true)
assert.equal(hasSystemConfig(undefined), false)
})
test('remoteUrlReady asks a new remote plugin for its service URL', () => {
const remote = (change: string) => ({ manifest: { runtime: { type: 'remote' } }, change })
assert.equal(remoteUrlReady(remote('install'), ''), false)
assert.equal(remoteUrlReady(remote('install'), 'plugins.example.com'), false)
assert.equal(remoteUrlReady(remote('install'), 'https://plugins.example.com'), true)
assert.equal(remoteUrlReady(remote('upgrade'), ''), true)
assert.equal(remoteUrlReady(remote('upgrade'), 'nope'), false)
assert.equal(remoteUrlReady({ manifest: { runtime: { type: 'host' } }, change: 'install' }, ''), true)
assert.equal(remoteUrlReady(null, ''), true)
})
@@ -114,6 +114,16 @@ export function installedState(p: InstalledPlugin): InstalledState {
export const EGRESS_ANY_HOST = '*'
/** Whether a string can be sent as a package URL. */
/**
* Whether the install drawer can go ahead with a remote plugin's service URL:
* a new install needs one, an upgrade may keep the registered one.
*/
export function remoteUrlReady(preview: { manifest: { runtime?: { type: string } }; change: string } | null, raw: string) {
if (preview?.manifest.runtime?.type !== 'remote') return true
if (raw.trim() === '') return preview.change !== 'install'
return isPackageUrl(raw)
}
export function isPackageUrl(raw: string): boolean {
try {
const u = new URL(raw.trim())
@@ -63,6 +63,34 @@
</ul>
</section>
<section v-if="plugin.runtime === 'remote'" class="setting-drawer__section">
<h4 class="setting-drawer__section-title">{{ t('pluginAdmin.detail.remote') }}</h4>
<div class="remote-row">
<span class="remote-row__label">{{ t('pluginAdmin.detail.remoteUrl') }}</span>
<template v-if="editingUrl">
<t-input v-model="urlDraft" size="small" class="remote-row__input" :disabled="savingUrl" @enter="saveUrl" />
<t-button size="small" theme="primary" :loading="savingUrl" :disabled="!isPackageUrl(urlDraft)" @click="saveUrl">
{{ t('common.save') }}
</t-button>
<t-button size="small" variant="text" :disabled="savingUrl" @click="editingUrl = false">
{{ t('common.cancel') }}
</t-button>
</template>
<template v-else>
<code class="remote-row__value">{{ plugin.remote_url }}</code>
<t-button size="small" variant="text" theme="primary" @click="startEditUrl">
{{ t('pluginAdmin.detail.editUrl') }}
</t-button>
</template>
</div>
<div class="danger-row">
<span class="form-desc">{{ t('pluginAdmin.detail.rotateHint') }}</span>
<t-popconfirm :content="t('pluginAdmin.detail.rotateConfirm')" @confirm="rotate">
<t-button variant="outline" :loading="rotating">{{ t('pluginAdmin.detail.rotateSecret') }}</t-button>
</t-popconfirm>
</div>
</section>
<section class="setting-drawer__section">
<h4 class="setting-drawer__section-title">{{ t('pluginAdmin.detail.versions') }}</h4>
<ul class="line-list">
@@ -119,6 +147,8 @@ import { getPlugin, type PluginInstance } from '@/api/plugin'
import {
activatePluginVersion,
getPluginSystemConfig,
rotatePluginSecret,
setPluginRemoteUrl,
uninstallPlugin,
updatePluginSystemConfig,
type InstalledPlugin,
@@ -130,6 +160,7 @@ import {
contributionLines,
formatBytes,
hasSystemConfig,
isPackageUrl,
shortDigest,
sortVersions,
} from '../pluginManagementState'
@@ -156,6 +187,10 @@ const configErrors = ref<FieldError[]>([])
const saving = ref(false)
const activating = ref('')
const uninstalling = ref(false)
const editingUrl = ref(false)
const urlDraft = ref('')
const savingUrl = ref(false)
const rotating = ref(false)
const formatDate = (s: string) => (s ? new Date(s).toLocaleString(locale.value) : '')
@@ -188,6 +223,7 @@ watch(
() => [props.visible, props.plugin?.id, props.plugin?.active_version, props.plugin?.desired_state] as const,
([visible, id]) => {
if (!visible || !id) return
editingUrl.value = false
void loadNodes(id)
void loadConfig(id)
},
@@ -226,6 +262,40 @@ async function activate(version: string) {
}
}
function startEditUrl() {
urlDraft.value = props.plugin?.remote_url ?? ''
editingUrl.value = true
}
async function saveUrl() {
if (!props.plugin || !isPackageUrl(urlDraft.value)) return
savingUrl.value = true
try {
const res = await setPluginRemoteUrl(props.plugin.id, urlDraft.value.trim())
editingUrl.value = false
emit('changed', res.data)
MessagePlugin.success(t('pluginAdmin.detail.urlSaved'))
} catch (e: any) {
MessagePlugin.error(e?.message || t('pluginAdmin.detail.urlSaveFailed'))
} finally {
savingUrl.value = false
}
}
// The new secret travels in the changed plugin; the page shows it once.
async function rotate() {
if (!props.plugin) return
rotating.value = true
try {
const res = await rotatePluginSecret(props.plugin.id)
emit('changed', res.data)
} catch (e: any) {
MessagePlugin.error(e?.message || t('pluginAdmin.detail.rotateFailed'))
} finally {
rotating.value = false
}
}
async function uninstall() {
if (!props.plugin) return
const id = props.plugin.id
@@ -318,6 +388,29 @@ async function uninstall() {
color: var(--td-text-color-placeholder);
}
.remote-row {
display: flex;
align-items: center;
flex-wrap: wrap;
gap: 8px;
font-size: var(--app-text-sm);
&__label {
color: var(--td-text-color-secondary);
}
&__value {
font-size: var(--app-text-xs);
color: var(--td-text-color-primary);
word-break: break-all;
}
&__input {
flex: 1;
min-width: 200px;
}
}
.danger-row {
display: flex;
align-items: center;
@@ -104,6 +104,17 @@
</ul>
</div>
<div v-if="isRemote" class="form-item">
<label class="form-label" :class="{ required: preview.change === 'install' }">
{{ t('pluginAdmin.install.remoteUrlLabel') }}
</label>
<t-input v-model="remoteUrl" :disabled="busy" placeholder="https://plugins.example.com/acme-search" />
<p class="form-desc">
{{ t('pluginAdmin.install.remoteUrlHint') }}
<template v-if="preview.change !== 'install'">{{ t('pluginAdmin.install.remoteUrlKeep') }}</template>
</p>
</div>
<t-alert
v-if="needsConfig"
theme="info"
@@ -139,6 +150,7 @@ import {
isPackageUrl,
permissionLines,
remoteHosts,
remoteUrlReady,
} from '../pluginManagementState'
import { hasTenantConfig } from '../../settings/pluginCenterState'
@@ -156,6 +168,7 @@ const { t, locale } = useI18n()
const mode = ref<'upload' | 'url'>('upload')
const file = ref<File | null>(null)
const url = ref('')
const remoteUrl = ref('')
const fileInput = ref<HTMLInputElement | null>(null)
const preview = ref<PluginPreview | null>(null)
const busy = ref(false)
@@ -167,6 +180,7 @@ watch(
mode.value = 'upload'
file.value = null
url.value = ''
remoteUrl.value = ''
preview.value = null
},
)
@@ -189,7 +203,10 @@ const source = computed<PackageSource | null>(() => {
return isPackageUrl(url.value) ? { url: url.value.trim() } : null
})
const canConfirm = computed(() => !!source.value && !busy.value)
const isRemote = computed(() => preview.value?.manifest.runtime?.type === 'remote')
const canConfirm = computed(
() => !!source.value && !busy.value && (!preview.value || remoteUrlReady(preview.value, remoteUrl.value)),
)
const confirmText = computed(() =>
preview.value ? t(`pluginAdmin.install.confirm.${preview.value.change}`) : t('pluginAdmin.install.inspect'),
)
@@ -237,7 +254,11 @@ async function onConfirm() {
if (!source.value) return
busy.value = true
try {
const res = await installPluginPackage(source.value, preview.value.digest)
const res = await installPluginPackage(
source.value,
preview.value.digest,
isRemote.value ? remoteUrl.value.trim() : undefined,
)
MessagePlugin.success(t('pluginAdmin.install.done', { name: localizedText(preview.value.manifest.name, locale.value) }))
emit('installed', res.data)
emit('update:visible', false)
+9 -5
View File
@@ -40,13 +40,17 @@ func (r *pluginRepository) GetPlugin(ctx context.Context, id string) (*types.Ins
}
// SavePlugin inserts or updates the row keyed by id.
// pluginUpdateColumns are what saving an installed plugin overwrites: every
// column but its identity and creation.
var pluginUpdateColumns = []string{
"owner_tenant_id", "source", "active_version", "desired_state", "runtime",
"granted_perms", "system_config", "remote_url", "remote_secret", "updated_at",
}
func (r *pluginRepository) SavePlugin(ctx context.Context, p *types.InstalledPlugin) error {
return r.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns([]string{
"owner_tenant_id", "source", "active_version", "desired_state", "runtime",
"granted_perms", "system_config", "updated_at",
}),
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns(pluginUpdateColumns),
}).Create(p).Error
}
@@ -2,6 +2,7 @@ package repository
import (
"context"
"sync"
"testing"
"time"
@@ -9,6 +10,7 @@ import (
"github.com/stretchr/testify/require"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"gorm.io/gorm/schema"
"github.com/Tencent/WeKnora/internal/types"
)
@@ -39,12 +41,15 @@ func TestPluginRepository(t *testing.T) {
require.NoError(t, repo.SavePlugin(ctx, p))
p.ActiveVersion = "1.1.0"
p.DesiredState = types.PluginStateDisabled
p.RemoteURL, p.RemoteSecret = "https://plugins.example.com", "enc:v1:x"
require.NoError(t, repo.SavePlugin(ctx, p), "save is an upsert")
got, err = repo.GetPlugin(ctx, "acme.kit")
require.NoError(t, err)
require.Equal(t, "1.1.0", got.ActiveVersion)
require.Equal(t, types.PluginStateDisabled, got.DesiredState)
require.Equal(t, "https://plugins.example.com", got.RemoteURL)
require.Equal(t, "enc:v1:x", got.RemoteSecret)
versions, err := repo.ListVersions(ctx, "acme.kit")
require.NoError(t, err)
@@ -142,3 +147,17 @@ func TestPluginKVRepository(t *testing.T) {
got, _ = repo.Get(ctx, "acme.x", 2, "cursor:a")
require.JSONEq(t, `99`, string(got.Value), "tenants are separate partitions")
}
// A column added to installed plugins must be saved on update too, or
// changing it silently does nothing.
func TestPluginUpdateColumnsCoverTheTable(t *testing.T) {
s, err := schema.Parse(&types.InstalledPlugin{}, &sync.Map{}, schema.NamingStrategy{})
require.NoError(t, err)
for _, name := range s.DBNames {
switch name {
case "id", "created_at", "created_by":
continue
}
require.Contains(t, pluginUpdateColumns, name, "SavePlugin does not update %s", name)
}
}
+7 -4
View File
@@ -78,6 +78,7 @@ import (
pluginmanifest "github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
pluginremote "github.com/Tencent/WeKnora/internal/plugin/remote"
plugintenancy "github.com/Tencent/WeKnora/internal/plugin/tenancy"
"github.com/Tencent/WeKnora/internal/router"
"github.com/Tencent/WeKnora/internal/sandbox"
@@ -168,6 +169,7 @@ func BuildContainer(container *dig.Container) *dig.Container {
must(container.Provide(activate.NewSkills))
must(container.Provide(activate.NewModelVendors))
must(container.Provide(pluginhost.NewManager))
must(container.Provide(pluginremote.NewManager))
must(container.Provide(newPluginInvoker))
must(container.Provide(repository.NewPluginKVRepository))
must(container.Provide(newPluginHostAPI))
@@ -1840,12 +1842,13 @@ func newPluginRegistry(
})
}
// newPluginDrivers returns the drivers that run plugins: builtins and
// declarative packages today; host, remote and kubernetes drivers join this
// set.
// newPluginDrivers returns the drivers that run plugins: builtins,
// declarative packages, the embedded host and remote services; a kubernetes
// driver joins this set.
func newPluginDrivers(reg *pluginregistry.Registry, r *reconcile.Reconciler) *plugindriver.Set {
return plugindriver.NewSet(plugindriver.NewBuiltin(reg.Plugin),
r.Driver(pluginmanifest.RuntimeDeclarative), r.Driver(pluginmanifest.RuntimeHost))
r.Driver(pluginmanifest.RuntimeDeclarative), r.Driver(pluginmanifest.RuntimeHost),
r.Driver(pluginmanifest.RuntimeRemote))
}
// installPluginGate gives the integration handlers the tenant plugin switches,
+35 -7
View File
@@ -21,8 +21,10 @@ import (
"github.com/Tencent/WeKnora/internal/plugin/install"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
"github.com/Tencent/WeKnora/internal/plugin/remote"
plugintenancy "github.com/Tencent/WeKnora/internal/plugin/tenancy"
"github.com/Tencent/WeKnora/internal/types/interfaces"
"github.com/Tencent/WeKnora/pluginsdk/client"
)
// newPluginPackageStore keeps plugin packages in the deployment's object
@@ -41,6 +43,7 @@ type pluginActivators struct {
dig.In
Host *host.Manager
Remote *remote.Manager
WebSearch *activate.WebSearch
Connectors *activate.Connectors
Parsers *activate.Parsers
@@ -50,19 +53,23 @@ type pluginActivators struct {
Invoker *activate.Invoker
}
// list orders the activators: the host first, so a code plugin's process is
// running before anything routes calls to it.
// list orders the activators: the runtimes first, so a code plugin is
// reachable before anything routes calls to it.
func (a pluginActivators) list() []reconcile.Activator {
return []reconcile.Activator{a.Host, a.WebSearch, a.Connectors, a.Parsers, a.Vendors, a.MCP, a.Skills}
return []reconcile.Activator{
a.Host, a.Remote, a.WebSearch, a.Connectors, a.Parsers, a.Vendors, a.MCP, a.Skills,
}
}
// newPluginHostAPI serves the Host API and gives calls a way back to it: the
// embedded host's plugins reach this node on loopback.
// embedded host's plugins reach this node on loopback, remote plugins only
// through WEKNORA_PLUGIN_HOST_API_URL.
func newPluginHostAPI(
cfg *config.Config, iv *activate.Invoker, repo interfaces.PluginKVRepository, cleaner interfaces.ResourceCleaner,
) *hostapi.Handler {
issuer := hostapi.NewIssuerFromEnv()
url := strings.TrimSpace(os.Getenv("WEKNORA_PLUGIN_HOST_API_URL"))
iv.SetPublicHostAPI(url)
if url == "" {
port := 8080
if cfg != nil && cfg.Server != nil && cfg.Server.Port > 0 {
@@ -90,8 +97,24 @@ func bindPluginActivators(
skills.SetPluginSkills(a.Skills)
}
// newPluginInvoker reaches code plugins through this node's plugin host.
func newPluginInvoker(h *host.Manager) *activate.Invoker { return activate.NewInvoker(h) }
// newPluginInvoker reaches code plugins through this node's plugin host or,
// for remote ones, at their registered URLs.
func newPluginInvoker(h *host.Manager, r *remote.Manager) *activate.Invoker {
return activate.NewInvoker(pluginClients{host: h, remote: r})
}
// pluginClients finds a code plugin in whichever runtime loaded it.
type pluginClients struct {
host *host.Manager
remote *remote.Manager
}
func (c pluginClients) Client(pluginID string) (*client.Client, error) {
if c.remote.Owns(pluginID) {
return c.remote.Client(pluginID)
}
return c.host.Client(pluginID)
}
func newMCPServiceRepository(db *gorm.DB, plugins *activate.MCPServers) interfaces.MCPServiceRepository {
return plugins.Repository(repository.NewMCPServiceRepository(db))
@@ -119,13 +142,18 @@ func newPluginInstaller(
// startPluginReconciler loads installed plugins before the server takes
// traffic, then keeps this node in step with the others.
func startPluginReconciler(r *reconcile.Reconciler, hostManager *host.Manager, cleaner interfaces.ResourceCleaner) {
func startPluginReconciler(
r *reconcile.Reconciler, hostManager *host.Manager, remoteManager *remote.Manager,
cleaner interfaces.ResourceCleaner,
) {
hostManager.SetReporter(r)
remoteManager.SetReporter(r)
ctx, cancel := context.WithCancel(context.Background())
r.Start(ctx)
cleaner.RegisterWithName("PluginReconciler", func() error {
cancel()
hostManager.Close()
remoteManager.Close()
return nil
})
}
@@ -73,7 +73,7 @@ var versionedSQLiteColumns = map[string][]string{
}, // 000028
}
const expectedSQLiteMigrationVersion = 36
const expectedSQLiteMigrationVersion = 37
func TestSQLiteMigrationsCreateVersionedSchema(t *testing.T) {
repoRoot := sqliteRepoRoot(t)
+79 -13
View File
@@ -32,23 +32,36 @@ type PluginPackageRequest struct {
URL string `json:"url"`
// Digest pins an install to the package reviewed with inspect.
Digest string `json:"digest"`
// RemoteURL is where a remote plugin's service runs.
RemoteURL string `json:"remote_url"`
}
// packageInput is a package with what its install request said about it.
type packageInput struct {
data []byte
source install.Source
digest string
remoteURL string
}
// readPackage returns the package archive from an upload or a URL.
func (h *PluginAdminHandler) readPackage(c *gin.Context) ([]byte, install.Source, string, bool) {
func (h *PluginAdminHandler) readPackage(c *gin.Context) (packageInput, bool) {
if strings.HasPrefix(c.ContentType(), "application/json") {
limitJSONBody(c, skillSourceJSONMaxBytes)
var req PluginPackageRequest
if err := c.ShouldBindJSON(&req); err != nil || strings.TrimSpace(req.URL) == "" {
_ = c.Error(errors.NewBadRequestError("url or an uploaded file is required"))
return nil, install.Source{}, "", false
return packageInput{}, false
}
data, err := h.service.FetchURL(c.Request.Context(), strings.TrimSpace(req.URL))
if err != nil {
h.fail(c, err)
return nil, install.Source{}, "", false
return packageInput{}, false
}
return data, install.Source{Kind: "url", URL: strings.TrimSpace(req.URL)}, req.Digest, true
return packageInput{
data: data, source: install.Source{Kind: "url", URL: strings.TrimSpace(req.URL)},
digest: req.Digest, remoteURL: strings.TrimSpace(req.RemoteURL),
}, true
}
limitUploadBody(c, pkg.MaxArchiveBytes)
@@ -59,19 +72,22 @@ func (h *PluginAdminHandler) readPackage(c *gin.Context) ([]byte, install.Source
} else {
_ = c.Error(errors.NewBadRequestError("file is required"))
}
return nil, install.Source{}, "", false
return packageInput{}, false
}
defer func() { _ = file.Close() }()
data, err := io.ReadAll(io.LimitReader(file, pkg.MaxArchiveBytes+1))
if err != nil {
_ = c.Error(errors.NewBadRequestError("failed to read the uploaded package"))
return nil, install.Source{}, "", false
return packageInput{}, false
}
if len(data) > pkg.MaxArchiveBytes {
_ = c.Error(errors.NewBadRequestError("plugin package is too large"))
return nil, install.Source{}, "", false
return packageInput{}, false
}
return data, install.Source{Kind: "upload", URL: header.Filename}, c.PostForm("digest"), true
return packageInput{
data: data, source: install.Source{Kind: "upload", URL: header.Filename},
digest: c.PostForm("digest"), remoteURL: strings.TrimSpace(c.PostForm("remote_url")),
}, true
}
func (h *PluginAdminHandler) fail(c *gin.Context, err error) {
@@ -101,11 +117,11 @@ func (h *PluginAdminHandler) ok(c *gin.Context, data any) {
// @Security Bearer
// @Router /system/admin/plugins/inspect [post]
func (h *PluginAdminHandler) InspectPlugin(c *gin.Context) {
data, _, _, ok := h.readPackage(c)
in, ok := h.readPackage(c)
if !ok {
return
}
preview, err := h.service.Inspect(c.Request.Context(), data)
preview, err := h.service.Inspect(c.Request.Context(), in.data)
if err != nil {
h.fail(c, err)
return
@@ -115,7 +131,9 @@ func (h *PluginAdminHandler) InspectPlugin(c *gin.Context) {
// InstallPlugin godoc
// @Summary 安装或升级插件
// @Description 安装插件包(multipart file + digest,或 JSON {url, digest})。digest 取自 inspect,保证安装的就是审阅过的包。安装后全平台可用,各空间需自行启用
// @Description 安装插件包(multipart file + digest,或 JSON {url, digest})。digest 取自 inspect,
// @Description 保证安装的就是审阅过的包。安装后全平台可用,各空间需自行启用。
// @Description remote 插件还需 remote_url(服务地址);首次安装的响应里 issuedSecret 是签名密钥,只返回这一次
// @Tags System
// @Accept multipart/form-data,json
// @Produce json
@@ -123,13 +141,13 @@ func (h *PluginAdminHandler) InspectPlugin(c *gin.Context) {
// @Security Bearer
// @Router /system/admin/plugins [post]
func (h *PluginAdminHandler) InstallPlugin(c *gin.Context) {
data, source, digest, ok := h.readPackage(c)
in, ok := h.readPackage(c)
if !ok {
return
}
userID, _ := c.Request.Context().Value(types.UserIDContextKey).(string)
view, err := h.service.Install(c.Request.Context(), install.Request{
Data: data, Source: source, ExpectedDigest: digest, UserID: userID,
Data: in.data, Source: in.source, ExpectedDigest: in.digest, UserID: userID, RemoteURL: in.remoteURL,
})
if err != nil {
h.fail(c, err)
@@ -227,6 +245,54 @@ func (h *PluginAdminHandler) ActivatePluginVersion(c *gin.Context) {
h.ok(c, view)
}
// SetPluginRemoteURLRequest moves a remote plugin.
type SetPluginRemoteURLRequest struct {
URL string `json:"url" binding:"required"`
}
// SetPluginRemoteURL godoc
// @Summary 修改远程插件的服务地址
// @Description 私有网络中的地址需要加入 SSRF_WHITELIST
// @Tags System
// @Accept json
// @Produce json
// @Param id path string true "插件 ID"
// @Param request body SetPluginRemoteURLRequest true "服务地址"
// @Success 200 {object} map[string]interface{}
// @Security Bearer
// @Router /system/admin/plugins/{id}/remote-url [put]
func (h *PluginAdminHandler) SetPluginRemoteURL(c *gin.Context) {
var req SetPluginRemoteURLRequest
if err := c.ShouldBindJSON(&req); err != nil {
_ = c.Error(errors.NewBadRequestError("url is required"))
return
}
view, err := h.service.SetRemoteURL(c.Request.Context(), c.Param("id"), strings.TrimSpace(req.URL))
if err != nil {
h.fail(c, err)
return
}
h.ok(c, view)
}
// RotatePluginSecret godoc
// @Summary 轮换远程插件的签名密钥
// @Description 新密钥只在本次响应的 issuedSecret 中返回;服务换上新密钥前调用会失败
// @Tags System
// @Produce json
// @Param id path string true "插件 ID"
// @Success 200 {object} map[string]interface{}
// @Security Bearer
// @Router /system/admin/plugins/{id}/secret/rotate [post]
func (h *PluginAdminHandler) RotatePluginSecret(c *gin.Context) {
view, err := h.service.RotateSecret(c.Request.Context(), c.Param("id"))
if err != nil {
h.fail(c, err)
return
}
h.ok(c, view)
}
// UninstallPlugin godoc
// @Summary 卸载插件
// @Description 删除插件及其全部版本;各空间的开关和配置保留,重新安装后恢复
+16 -3
View File
@@ -28,9 +28,11 @@ type Invoker struct {
mu sync.RWMutex
tenancy *tenancy.Service
plugins interfaces.PluginRepository
// tokens and hostURL give plugins granted Host API scopes a way back.
tokens TokenIssuer
hostURL string
// tokens and hostURL give plugins granted Host API scopes a way back;
// remote plugins, off this node, use publicHostURL.
tokens TokenIssuer
hostURL string
publicHostURL string
}
// TokenIssuer signs the Host API token of one call.
@@ -46,6 +48,14 @@ func (iv *Invoker) SetHostAPI(tokens TokenIssuer, url string) {
iv.mu.Unlock()
}
// SetPublicHostAPI is where remote plugins reach the Host API. Without it
// their calls carry no Host API access.
func (iv *Invoker) SetPublicHostAPI(url string) {
iv.mu.Lock()
iv.publicHostURL = url
iv.mu.Unlock()
}
// NewInvoker creates an Invoker; Bind completes it.
func NewInvoker(clients ClientSource) *Invoker { return &Invoker{clients: clients} }
@@ -94,6 +104,9 @@ func (iv *Invoker) Envelope(
}
iv.mu.RLock()
t, plugins, tokens, hostURL := iv.tenancy, iv.plugins, iv.tokens, iv.hostURL
if m.Runtime.Type == manifest.RuntimeRemote {
hostURL = iv.publicHostURL
}
iv.mu.RUnlock()
if tokens != nil && hostURL != "" && env.Context.TenantID != 0 && len(m.Permissions.HostAPI) > 0 {
token, _, err := tokens.Issue(m.ID, m.Version, env.Context.TenantID, m.Permissions.HostAPI)
+11
View File
@@ -34,4 +34,15 @@ func TestEnvelopeCarriesHostAccessOnlyForGrantedPlugins(t *testing.T) {
if env, _ := iv.Envelope(context.Background(), granted, nil); env.Context.Host != nil {
t.Fatal("a call without a tenant must get no token")
}
remote := *granted
remote.Runtime.Type = manifest.RuntimeRemote
if env, _ := iv.Envelope(ctx, &remote, nil); env.Context.Host != nil {
t.Fatal("a remote plugin must not be sent to this node's loopback address")
}
iv.SetPublicHostAPI("https://weknora.example.com")
if env, _ := iv.Envelope(ctx, &remote, nil); env.Context.Host == nil ||
env.Context.Host.URL != "https://weknora.example.com" {
t.Fatalf("remote envelope = %+v", env.Context)
}
}
+6 -6
View File
@@ -248,7 +248,7 @@ func (p *process) launch(ctx context.Context, entry string) (*launched, error) {
kill()
return nil, fmt.Errorf("read plugin manifest: %w", err)
}
if err := checkServedManifest(p.spec.m, m); err != nil {
if err := CheckServedManifest(p.spec.m, m); err != nil {
c.Close()
kill()
return nil, err
@@ -256,14 +256,14 @@ func (p *process) launch(ctx context.Context, entry string) (*launched, error) {
return &launched{cmd: cmd, client: c, exited: exited}, nil
}
// checkServedManifest makes sure the process is the package that was
// CheckServedManifest makes sure a running plugin is the package that was
// installed and serves what the manifest promised.
func checkServedManifest(want *manifest.Manifest, got *pluginapi.Manifest) error {
func CheckServedManifest(want *manifest.Manifest, got *pluginapi.Manifest) error {
if got.ID != want.ID || got.Version != want.Version {
return fmt.Errorf("process reports %s@%s, the package is %s@%s", got.ID, got.Version, want.ID, want.Version)
return fmt.Errorf("plugin reports %s@%s, the package is %s@%s", got.ID, got.Version, want.ID, want.Version)
}
if got.APIVersion != pluginapi.APIVersion {
return fmt.Errorf("process speaks %q, this WeKnora speaks %q", got.APIVersion, pluginapi.APIVersion)
return fmt.Errorf("plugin speaks %q, this WeKnora speaks %q", got.APIVersion, pluginapi.APIVersion)
}
for point, contribs := range want.Contributes {
if info, ok := manifest.LookupPoint(point); !ok || info.Declarative {
@@ -275,7 +275,7 @@ func checkServedManifest(want *manifest.Manifest, got *pluginapi.Manifest) error
}
for _, c := range contribs {
if !served[c.ID] {
return fmt.Errorf("plugin.yaml declares %s/%s but the process does not serve it", point, c.ID)
return fmt.Errorf("plugin.yaml declares %s/%s but the plugin does not serve it", point, c.ID)
}
}
}
+115 -2
View File
@@ -6,6 +6,8 @@ package install
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -47,6 +49,7 @@ func invalid(format string, args ...any) error {
var supportedRuntimes = map[manifest.RuntimeType]bool{
manifest.RuntimeDeclarative: true,
manifest.RuntimeHost: true,
manifest.RuntimeRemote: true,
}
// checkHostRuntime makes sure this server can run a host plugin: a binary
@@ -142,6 +145,9 @@ type View struct {
Versions []types.PluginVersion `json:"versions"`
// Node is the plugin's state on the node that served the request.
Node *reconcile.Status `json:"node,omitempty"`
// IssuedSecret is a remote plugin's signing secret, returned only by
// the call that created it: the service needs it to check requests.
IssuedSecret string `json:"issuedSecret,omitempty"`
}
// Inspect opens a package and says what installing it would do.
@@ -176,7 +182,7 @@ func (s *Service) open(data []byte) (*pkg.Package, error) {
return nil, &InvalidError{Err: err}
}
if !supportedRuntimes[p.Manifest.Runtime.Type] {
return nil, invalid("runtime %q is not supported yet; declarative and host plugins can be installed",
return nil, invalid("runtime %q is not supported yet; declarative, host and remote plugins can be installed",
p.Manifest.Runtime.Type)
}
if p.Manifest.Runtime.Type == manifest.RuntimeHost {
@@ -203,6 +209,9 @@ type Request struct {
// to the package the administrator reviewed.
ExpectedDigest string
UserID string
// RemoteURL is where a remote plugin's service runs. An upgrade may
// leave it empty to keep the registered one.
RemoteURL string
}
// Install stores the package as a version of its plugin and makes it the
@@ -248,6 +257,23 @@ func (s *Service) Install(ctx context.Context, req Request) (*View, error) {
if row == nil {
row = &types.InstalledPlugin{ID: m.ID, DesiredState: types.PluginStateEnabled, CreatedBy: req.UserID}
}
var issued string
if m.Runtime.Type == manifest.RuntimeRemote {
if req.RemoteURL != "" {
if err := checkRemoteURL(req.RemoteURL); err != nil {
return nil, err
}
row.RemoteURL = strings.TrimSuffix(req.RemoteURL, "/")
}
if row.RemoteURL == "" {
return nil, invalid("a remote plugin needs the URL of its service")
}
if row.RemoteSecret == "" {
if issued, err = s.issueSecret(row); err != nil {
return nil, err
}
}
}
source, _ := json.Marshal(req.Source)
perms, _ := json.Marshal(m.Permissions)
row.Source = types.JSON(source)
@@ -258,7 +284,94 @@ func (s *Service) Install(ctx context.Context, req Request) (*View, error) {
return nil, err
}
logger.Infof(ctx, "[plugin] %s installed %s %s (%s)", req.UserID, m.ID, m.Version, p.Digest)
return s.apply(ctx, m.ID)
v, err := s.apply(ctx, m.ID)
if v != nil {
v.IssuedSecret = issued
}
return v, err
}
// checkRemoteURL accepts an http(s) service URL that passes the SSRF rules;
// a service on a private network needs its host in SSRF_WHITELIST.
func checkRemoteURL(raw string) error {
u, err := url.Parse(raw)
if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" {
return invalid("service URL must be an http(s) URL")
}
if u.User != nil || u.RawQuery != "" || u.Fragment != "" {
return invalid("service URL must not carry credentials, a query or a fragment")
}
if err := utils.ValidateURLForSSRF(raw); err != nil {
return invalid("service URL is not allowed (private hosts must be in SSRF_WHITELIST): %v", err)
}
return nil
}
// issueSecret gives a remote plugin a new signing secret, stored sealed,
// and returns it in the clear for the administrator to hand to the service.
func (s *Service) issueSecret(row *types.InstalledPlugin) (string, error) {
b := make([]byte, 32)
if _, err := rand.Read(b); err != nil {
return "", err
}
secret := hex.EncodeToString(b)
sealed, err := utils.EncryptAESGCM(secret, utils.GetAESKey())
if err != nil {
return "", err
}
row.RemoteSecret = sealed
return secret, nil
}
// SetRemoteURL moves a remote plugin to another service URL.
func (s *Service) SetRemoteURL(ctx context.Context, id, rawURL string) (*View, error) {
row, err := s.remoteRow(ctx, id)
if err != nil {
return nil, err
}
if err := checkRemoteURL(rawURL); err != nil {
return nil, err
}
row.RemoteURL = strings.TrimSuffix(rawURL, "/")
if err := s.repo.SavePlugin(ctx, row); err != nil {
return nil, err
}
return s.apply(ctx, id)
}
// RotateSecret replaces a remote plugin's signing secret. Calls fail until
// the service is given the new one, which only this response shows.
func (s *Service) RotateSecret(ctx context.Context, id string) (*View, error) {
row, err := s.remoteRow(ctx, id)
if err != nil {
return nil, err
}
secret, err := s.issueSecret(row)
if err != nil {
return nil, err
}
if err := s.repo.SavePlugin(ctx, row); err != nil {
return nil, err
}
v, err := s.apply(ctx, id)
if v != nil {
v.IssuedSecret = secret
}
return v, err
}
func (s *Service) remoteRow(ctx context.Context, id string) (*types.InstalledPlugin, error) {
row, err := s.repo.GetPlugin(ctx, id)
if err != nil {
return nil, err
}
if row == nil {
return nil, ErrNotInstalled
}
if row.Runtime != string(manifest.RuntimeRemote) {
return nil, invalid("%s is not a remote plugin", id)
}
return row, nil
}
// FetchURL downloads a package over HTTP(S), refusing private addresses.
+72
View File
@@ -11,6 +11,7 @@ import (
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
"github.com/Tencent/WeKnora/internal/plugin/registry"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/utils"
)
func newService(t *testing.T) (*Service, *plugintest.MemRepo, *plugintest.MemStore, *registry.Registry) {
@@ -161,3 +162,74 @@ func TestInstallHostPlugins(t *testing.T) {
t.Fatalf("want an unsupported kind error, got %v", err)
}
}
func remotePackage(t *testing.T, version string) []byte {
return plugintest.Zip(t, map[string]string{
"plugin.yaml": "schemaVersion: 1\nid: acme.remote\nversion: " + version +
"\napiVersion: weknora.plugin/v1\n" +
"name: { en-US: ACME Remote }\npublisher: { id: acme }\nruntime: { type: remote }\n" +
"contributes:\n webSearch:\n - { id: search, name: ACME Search }\n",
})
}
func TestInstallRemotePlugins(t *testing.T) {
ctx := context.Background()
utils.SetSSRFWhitelistFromRaw("plugins.example.com")
t.Cleanup(func() { utils.SetSSRFWhitelistFromRaw("") })
t.Setenv("SYSTEM_AES_KEY", strings.Repeat("k", 32))
s, repo, _, _ := newService(t)
if _, err := s.Install(ctx, Request{Data: remotePackage(t, "1.0.0")}); !isInvalid(err) ||
!strings.Contains(err.Error(), "URL of its service") {
t.Fatalf("want a missing URL error, got %v", err)
}
refused := []string{"http://127.0.0.1:9000", "ftp://plugins.example.com", "https://u:p@plugins.example.com"}
for _, u := range refused {
if _, err := s.Install(ctx, Request{Data: remotePackage(t, "1.0.0"), RemoteURL: u}); !isInvalid(err) {
t.Errorf("RemoteURL %s = %v, want refusal", u, err)
}
}
view, err := s.Install(ctx, Request{
Data: remotePackage(t, "1.0.0"), RemoteURL: "https://plugins.example.com/acme/",
})
if err != nil {
t.Fatal(err)
}
if len(view.IssuedSecret) != 64 || view.RemoteURL != "https://plugins.example.com/acme" {
t.Fatalf("view = %+v", view)
}
row, _ := repo.GetPlugin(ctx, "acme.remote")
if plain, _ := utils.DecryptStoredSecret(row.RemoteSecret); !strings.HasPrefix(row.RemoteSecret, utils.EncPrefix) ||
plain != view.IssuedSecret {
t.Fatalf("stored secret %q does not seal the issued one", row.RemoteSecret)
}
first := view.IssuedSecret
view, err = s.Install(ctx, Request{Data: remotePackage(t, "1.1.0")})
if err != nil || view.IssuedSecret != "" || view.RemoteURL != "https://plugins.example.com/acme" {
t.Fatalf("upgrade must keep the URL and secret: %+v, %v", view, err)
}
view, err = s.RotateSecret(ctx, "acme.remote")
if err != nil || len(view.IssuedSecret) != 64 || view.IssuedSecret == first {
t.Fatalf("rotate = %+v, %v", view, err)
}
if view, err = s.SetRemoteURL(ctx, "acme.remote", "https://plugins.example.com/v2"); err != nil ||
view.RemoteURL != "https://plugins.example.com/v2" || view.IssuedSecret != "" {
t.Fatalf("move = %+v, %v", view, err)
}
if _, err := s.SetRemoteURL(ctx, "acme.remote", "http://10.0.0.1"); !isInvalid(err) {
t.Fatalf("private URL = %v", err)
}
if _, err := s.Install(ctx, Request{Data: plugintest.KitPackage(t, "1.0.0")}); err != nil {
t.Fatal(err)
}
if _, err := s.RotateSecret(ctx, "acme.kit"); !isInvalid(err) {
t.Fatalf("rotating a declarative plugin = %v", err)
}
if _, err := s.RotateSecret(ctx, "acme.none"); !errors.Is(err, ErrNotInstalled) {
t.Fatalf("rotating a missing plugin = %v", err)
}
}
+64 -8
View File
@@ -8,6 +8,8 @@ package reconcile
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -37,6 +39,9 @@ type Loaded struct {
// Dir is the package extracted on local disk, for consumers that read
// files by path (skills).
Dir string
// Installed is the plugin's row: where a remote plugin runs and its
// sealed secret.
Installed types.InstalledPlugin
}
// Activator wires one domain to plugin contributions: it registers what a
@@ -101,14 +106,40 @@ type Reconciler struct {
instanceID string
interval time.Duration
mu sync.Mutex // serializes passes
loaded map[string]*Loaded
digests map[string]string // plugin ID → loaded digest
mu sync.Mutex // serializes passes
loaded map[string]*Loaded
digests map[string]string // plugin ID → loaded digest and runtime target
// retries holds plugins whose activation failed: a process that would
// not start, a remote service that was down. They are tried again with
// backoff until they load or change.
retries map[string]retry
now func() time.Time
statusMu sync.RWMutex
status map[string]Status
runOnce sync.Once
}
// retry is when a failed activation is tried again.
type retry struct {
key string // the load key that failed
attempts int
next time.Time
}
// Backoff between attempts to activate a plugin that failed.
const (
retryFloor = 30 * time.Second
retryCeiling = 10 * time.Minute
)
func retryDelay(attempts int) time.Duration {
d := retryFloor
for i := 1; i < attempts && d < retryCeiling; i++ {
d *= 2
}
return min(d, retryCeiling)
}
// Options configures a Reconciler.
type Options struct {
Repo interfaces.PluginRepository
@@ -131,7 +162,8 @@ func New(o Options) *Reconciler {
return &Reconciler{
repo: o.Repo, store: o.Store, registry: o.Registry, cacheDir: o.CacheDir, rdb: o.Redis,
activators: o.Activators, instanceID: uuid.NewString(), interval: o.Interval,
loaded: map[string]*Loaded{}, digests: map[string]string{}, status: map[string]Status{},
loaded: map[string]*Loaded{}, digests: map[string]string{}, retries: map[string]retry{},
status: map[string]Status{}, now: time.Now,
}
}
@@ -165,7 +197,7 @@ func (r *Reconciler) Reconcile(ctx context.Context) error {
r.setStatus(row.ID, Status{Version: row.ActiveVersion, State: StateFailed, Error: err.Error()})
}
}
for id := range r.digests {
for id := range r.loaded {
if !desired[id] {
r.unload(ctx, id)
}
@@ -183,6 +215,16 @@ func (r *Reconciler) Reconcile(ctx context.Context) error {
return errors.Join(errs...)
}
// runtimeTarget identifies where a plugin runs beyond its package: a new
// remote URL or secret reloads the plugin like a new version would.
func runtimeTarget(row types.InstalledPlugin) string {
if row.RemoteURL == "" && row.RemoteSecret == "" {
return ""
}
sum := sha256.Sum256([]byte(row.RemoteURL + "\x00" + row.RemoteSecret))
return hex.EncodeToString(sum[:8])
}
// ensure loads the active version of one plugin unless it already is.
func (r *Reconciler) ensure(ctx context.Context, row types.InstalledPlugin) error {
v, err := r.repo.GetVersion(ctx, row.ID, row.ActiveVersion)
@@ -192,9 +234,13 @@ func (r *Reconciler) ensure(ctx context.Context, row types.InstalledPlugin) erro
if v == nil {
return fmt.Errorf("version %s is not stored", row.ActiveVersion)
}
if r.digests[row.ID] == v.Digest {
loadKey := v.Digest + "|" + runtimeTarget(row)
if r.digests[row.ID] == loadKey {
return nil
}
if rt, ok := r.retries[row.ID]; ok && rt.key == loadKey && r.now().Before(rt.next) {
return nil // still failed; its status says why
}
data, err := r.store.Get(ctx, v.PackageURI)
if err != nil {
return err
@@ -216,7 +262,7 @@ func (r *Reconciler) ensure(ctx context.Context, row types.InstalledPlugin) erro
if err := r.registry.Replace(p.Manifest); err != nil {
return err
}
l := &Loaded{Manifest: p.Manifest, Package: p, Dir: dir}
l := &Loaded{Manifest: p.Manifest, Package: p, Dir: dir, Installed: row}
var errs []error
for _, a := range r.activators {
_, inPlace := a.(InPlaceActivator)
@@ -230,10 +276,19 @@ func (r *Reconciler) ensure(ctx context.Context, row types.InstalledPlugin) erro
}
}
r.loaded[row.ID] = l
r.digests[row.ID] = v.Digest
if err := errors.Join(errs...); err != nil {
rt := r.retries[row.ID]
if rt.key != loadKey {
rt = retry{key: loadKey}
}
rt.attempts++
rt.next = r.now().Add(retryDelay(rt.attempts))
r.retries[row.ID] = rt
delete(r.digests, row.ID)
return err
}
r.digests[row.ID] = loadKey
delete(r.retries, row.ID)
logger.Infof(ctx, "[plugin] loaded %s %s", row.ID, row.ActiveVersion)
r.setStatus(row.ID, Status{Version: row.ActiveVersion, State: StateReady})
return nil
@@ -252,6 +307,7 @@ func (r *Reconciler) unload(ctx context.Context, id string) {
}
delete(r.loaded, id)
delete(r.digests, id)
delete(r.retries, id)
r.statusMu.Lock()
delete(r.status, id)
r.statusMu.Unlock()
@@ -7,6 +7,7 @@ import (
"path/filepath"
"strings"
"testing"
"time"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/plugin/plugintest"
@@ -128,3 +129,100 @@ func TestActivatorFailureMarksPluginFailed(t *testing.T) {
t.Fatalf("status = %+v", s)
}
}
// A plugin whose activation failed (a service that was down) is tried again
// with backoff, and loads once the cause is gone.
func TestFailedActivationIsRetriedWithBackoff(t *testing.T) {
ctx := context.Background()
repo, store, act := plugintest.NewMemRepo(), &plugintest.MemStore{}, &recorder{fail: true}
r := New(Options{
Repo: repo, Store: store, Registry: registry.New(), CacheDir: t.TempDir(), Activators: []Activator{act},
})
now := time.Unix(1000, 0)
r.now = func() time.Time { return now }
plugintest.Install(t, repo, store, plugintest.KitPackage(t, "1.0.0"), types.PluginStateEnabled)
if err := r.Reconcile(ctx); err == nil {
t.Fatal("want the activation error")
}
now = now.Add(retryFloor - time.Second)
_ = r.Reconcile(ctx)
if len(act.calls) != 1 {
t.Fatalf("retried before the backoff: %v", act.calls)
}
now = now.Add(time.Second)
_ = r.Reconcile(ctx)
if len(act.calls) != 3 { // deactivate the half-loaded plugin, activate again
t.Fatalf("calls = %v", act.calls)
}
now = now.Add(retryFloor)
_ = r.Reconcile(ctx)
if len(act.calls) != 3 {
t.Fatalf("the second retry should wait twice as long: %v", act.calls)
}
act.fail = false
now = now.Add(retryFloor)
if err := r.Reconcile(ctx); err != nil {
t.Fatal(err)
}
if s, _ := r.Status("acme.kit"); s.State != StateReady {
t.Fatalf("status = %+v", s)
}
_ = r.Reconcile(ctx)
if len(act.calls) != 5 {
t.Fatalf("a loaded plugin should stay loaded: %v", act.calls)
}
// A failed plugin that is disabled is unloaded all the same.
act.fail = true
plugintest.Install(t, repo, store, plugintest.KitPackage(t, "1.1.0"), types.PluginStateEnabled)
_ = r.Reconcile(ctx)
row, _ := repo.GetPlugin(ctx, "acme.kit")
row.DesiredState = types.PluginStateDisabled
_ = repo.SavePlugin(ctx, row)
_ = r.Reconcile(ctx)
if len(r.Loaded()) != 0 {
t.Fatal("a disabled plugin must be unloaded even if it failed")
}
}
func TestRetryDelay(t *testing.T) {
if retryDelay(1) != retryFloor || retryDelay(2) != 2*retryFloor || retryDelay(50) != retryCeiling {
t.Fatalf("delays = %s %s %s", retryDelay(1), retryDelay(2), retryDelay(50))
}
}
// A remote plugin moved to another URL, or given a new secret, is loaded
// again so its runtime picks the change up.
func TestReconcileReloadsOnRuntimeTargetChange(t *testing.T) {
ctx := context.Background()
repo, store, reg, act := plugintest.NewMemRepo(), &plugintest.MemStore{}, registry.New(), &recorder{}
r := New(Options{Repo: repo, Store: store, Registry: reg, CacheDir: t.TempDir(), Activators: []Activator{act}})
plugintest.Install(t, repo, store, plugintest.KitPackage(t, "1.0.0"), types.PluginStateEnabled)
row, _ := repo.GetPlugin(ctx, "acme.kit")
row.RemoteURL, row.RemoteSecret = "https://a.example.com", "s1"
_ = repo.SavePlugin(ctx, row)
if err := r.Reconcile(ctx); err != nil {
t.Fatal(err)
}
if got := r.Loaded()[0].Installed.RemoteURL; got != "https://a.example.com" {
t.Fatalf("Loaded carries URL %q", got)
}
_ = r.Reconcile(ctx)
row.RemoteURL = "https://b.example.com"
_ = repo.SavePlugin(ctx, row)
_ = r.Reconcile(ctx)
row.RemoteSecret = "s2"
_ = repo.SavePlugin(ctx, row)
_ = r.Reconcile(ctx)
if len(act.calls) != 5 || act.calls[1] != "deactivate acme.kit" {
t.Fatalf("calls = %v, want a reload per change", act.calls)
}
if got := r.Loaded()[0].Installed.RemoteSecret; got != "s2" {
t.Fatalf("Loaded carries secret %q", got)
}
}
+248
View File
@@ -0,0 +1,248 @@
// Package remote reaches remote-runtime plugins: HTTP services an
// administrator runs somewhere else and registers by URL. Every request is
// signed with the plugin's shared secret, so the service can tell WeKnora
// from anyone else who can reach it.
package remote
import (
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/plugin/host"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
"github.com/Tencent/WeKnora/internal/utils"
"github.com/Tencent/WeKnora/pluginsdk/client"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
const (
healthInterval = 15 * time.Second
checkTimeout = 10 * time.Second
)
// Manager keeps a client for every loaded remote plugin and health-checks
// it. It is a reconcile.Activator and, like the embedded host, must come
// before the activators that route calls to the plugins.
type Manager struct {
newClient func(url string, secret []byte) *client.Client
interval time.Duration
mu sync.Mutex
reporter host.StateReporter
endpoints map[string]*endpoint
}
// endpoint is one registered remote plugin.
type endpoint struct {
m *manifest.Manifest
c *client.Client
cancel context.CancelFunc
done chan struct{}
mu sync.Mutex
// err is why the service cannot be called right now: unreachable,
// or serving something other than the installed package.
err error
}
// NewManager creates a Manager whose clients refuse private addresses
// unless SSRF_WHITELIST allows them.
func NewManager() *Manager {
cfg := utils.DefaultSSRFSafeHTTPClientConfig()
cfg.Timeout = 0 // calls are bounded by their context; syncs stream
cfg.SameOriginRedirectsOnly = true
httpClient := utils.NewSSRFSafeHTTPClient(cfg)
return &Manager{
newClient: func(url string, secret []byte) *client.Client {
return client.New(url, httpClient, client.Signed(secret))
},
interval: healthInterval,
endpoints: map[string]*endpoint{},
}
}
// SetReporter wires health changes to the reconciler.
func (m *Manager) SetReporter(r host.StateReporter) {
m.mu.Lock()
m.reporter = r
m.mu.Unlock()
}
// Name implements reconcile.Activator.
func (m *Manager) Name() string { return "remote" }
// ActivatesInPlace implements reconcile.InPlaceActivator: a new URL or
// version takes over from the old endpoint without a gap.
func (m *Manager) ActivatesInPlace() {}
// Activate checks that the registered service is up and serves the
// installed package, then keeps health-checking it. Other runtimes are
// ignored.
func (m *Manager) Activate(ctx context.Context, l *reconcile.Loaded) error {
if l.Manifest.Runtime.Type != manifest.RuntimeRemote {
return nil
}
url := l.Installed.RemoteURL
if url == "" {
return errors.New("no service URL is registered for this remote plugin")
}
if err := utils.ValidateURLForSSRF(url); err != nil {
return fmt.Errorf("service URL is not allowed: %w", err)
}
secret, err := utils.DecryptStoredSecret(l.Installed.RemoteSecret)
if err != nil {
return fmt.Errorf("decrypt the plugin secret: %w", err)
}
if secret == "" {
return errors.New("the remote plugin has no signing secret")
}
c := m.newClient(url, []byte(secret))
cctx, cancel := context.WithTimeout(ctx, checkTimeout)
err = verify(cctx, c, l.Manifest)
cancel()
if err != nil {
c.Close()
return err
}
id := l.Manifest.ID
wctx, stop := context.WithCancel(context.Background())
e := &endpoint{m: l.Manifest, c: c, cancel: stop, done: make(chan struct{})}
m.mu.Lock()
old := m.endpoints[id]
m.endpoints[id] = e
m.mu.Unlock()
if old != nil {
old.close()
}
go m.watch(wctx, id, e)
logger.Infof(ctx, "[plugin] remote %s %s at %s", id, l.Manifest.Version, url)
return nil
}
// verify checks that the service is healthy and is the installed package.
func verify(ctx context.Context, c *client.Client, want *manifest.Manifest) error {
if err := c.Health(ctx); err != nil {
return fmt.Errorf("plugin service is not healthy: %w", err)
}
got, err := c.Manifest(ctx)
if err != nil {
return fmt.Errorf("read plugin manifest: %w", err)
}
return host.CheckServedManifest(want, got)
}
// watch health-checks the service until the endpoint is replaced or
// removed. After an outage it checks the manifest again: the service may
// have been redeployed with another version.
func (m *Manager) watch(ctx context.Context, id string, e *endpoint) {
defer close(e.done)
t := time.NewTicker(m.interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
}
cctx, cancel := context.WithTimeout(ctx, checkTimeout)
var err error
if e.failing() {
err = verify(cctx, e.c, e.m)
} else if err = e.c.Health(cctx); err != nil {
err = fmt.Errorf("plugin service is not healthy: %w", err)
}
cancel()
if ctx.Err() != nil {
return
}
if e.setErr(err) {
m.report(id, err)
}
}
}
func (m *Manager) report(id string, err error) {
m.mu.Lock()
r := m.reporter
m.mu.Unlock()
if r != nil {
r.ReportRuntime(id, err == nil, err)
}
}
func (e *endpoint) failing() bool {
e.mu.Lock()
defer e.mu.Unlock()
return e.err != nil
}
// setErr records the latest check and says whether health changed.
func (e *endpoint) setErr(err error) bool {
e.mu.Lock()
defer e.mu.Unlock()
changed := (e.err == nil) != (err == nil)
e.err = err
return changed
}
func (e *endpoint) close() {
e.cancel()
<-e.done
e.c.Close()
}
// Deactivate implements reconcile.Activator.
func (m *Manager) Deactivate(ctx context.Context, pluginID string) error {
m.mu.Lock()
e := m.endpoints[pluginID]
delete(m.endpoints, pluginID)
m.mu.Unlock()
if e != nil {
e.close()
logger.Infof(ctx, "[plugin] remote %s unregistered", pluginID)
}
return nil
}
// Client returns the client of a registered remote plugin. The error is a
// retryable pluginapi unavailable error while the service fails its checks.
func (m *Manager) Client(pluginID string) (*client.Client, error) {
m.mu.Lock()
e := m.endpoints[pluginID]
m.mu.Unlock()
if e == nil {
return nil, pluginapi.Errorf(pluginapi.CodeUnavailable, "remote plugin %s is not registered", pluginID)
}
e.mu.Lock()
err := e.err
e.mu.Unlock()
if err != nil {
return nil, pluginapi.Errorf(pluginapi.CodeUnavailable, "%v", err)
}
return e.c, nil
}
// Owns says whether a plugin is registered here.
func (m *Manager) Owns(pluginID string) bool {
m.mu.Lock()
defer m.mu.Unlock()
_, ok := m.endpoints[pluginID]
return ok
}
// Close unregisters every plugin, for shutdown.
func (m *Manager) Close() {
m.mu.Lock()
eps := m.endpoints
m.endpoints = map[string]*endpoint{}
m.mu.Unlock()
for _, e := range eps {
e.close()
}
}
+196
View File
@@ -0,0 +1,196 @@
package remote
import (
"bytes"
"context"
"errors"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/utils"
"github.com/Tencent/WeKnora/pluginsdk"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// service is a remote plugin that checks WeKnora's signature and can be
// taken down.
type service struct {
*httptest.Server
down atomic.Bool
}
func newService(t *testing.T, version, secret string) *service {
t.Helper()
p := pluginsdk.New(pluginsdk.Info{ID: "acme.search", Version: version})
p.WebSearch("web", pluginsdk.WebSearchFunc(
func(context.Context, *pluginsdk.Call, pluginapi.SearchInput) (*pluginapi.SearchOutput, error) {
return &pluginapi.SearchOutput{}, nil
}))
h := p.Handler()
s := &service{}
s.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if s.down.Load() {
http.Error(w, "down", http.StatusBadGateway)
return
}
body, _ := io.ReadAll(r.Body)
err := pluginapi.VerifySignature([]byte(secret), r.Header.Get(pluginapi.TimestampHeader),
r.Header.Get(pluginapi.SignatureHeader), body, time.Now())
if err != nil {
http.Error(w, err.Error(), http.StatusUnauthorized)
return
}
r.Body = io.NopCloser(bytes.NewReader(body))
h.ServeHTTP(w, r)
}))
t.Cleanup(s.Close)
return s
}
type reports struct {
mu sync.Mutex
got []bool
}
func (r *reports) ReportRuntime(_ string, healthy bool, _ error) {
r.mu.Lock()
r.got = append(r.got, healthy)
r.mu.Unlock()
}
func (r *reports) seen() []bool {
r.mu.Lock()
defer r.mu.Unlock()
return append([]bool(nil), r.got...)
}
func loaded(version, url, secret string) *reconcile.Loaded {
return &reconcile.Loaded{
Manifest: &manifest.Manifest{
ID: "acme.search", Version: version, APIVersion: manifest.ExtensionAPIVersion,
Runtime: manifest.Runtime{Type: manifest.RuntimeRemote},
Contributes: map[manifest.Point][]manifest.Contribution{
manifest.PointWebSearch: {{ID: "web"}},
},
},
Installed: types.InstalledPlugin{ID: "acme.search", RemoteURL: url, RemoteSecret: secret},
}
}
func allowLoopback(t *testing.T) {
utils.SetSSRFWhitelistFromRaw("127.0.0.1")
t.Cleanup(func() { utils.SetSSRFWhitelistFromRaw("") })
}
func TestActivateVerifiesAndSignsCalls(t *testing.T) {
allowLoopback(t)
ctx := context.Background()
svc := newService(t, "1.0.0", "s3cret")
m := NewManager()
defer m.Close()
if err := m.Activate(ctx, loaded("1.0.0", svc.URL, "wrong")); err == nil ||
!strings.Contains(err.Error(), "not healthy") {
t.Fatalf("wrong secret = %v", err)
}
if err := m.Activate(ctx, loaded("2.0.0", svc.URL, "s3cret")); err == nil ||
!strings.Contains(err.Error(), "the package is acme.search@2.0.0") {
t.Fatalf("other version = %v", err)
}
if err := m.Activate(ctx, loaded("1.0.0", "", "s3cret")); err == nil {
t.Fatal("no URL should fail")
}
if err := m.Activate(ctx, loaded("1.0.0", svc.URL, "s3cret")); err != nil {
t.Fatal(err)
}
c, err := m.Client("acme.search")
if err != nil {
t.Fatal(err)
}
var out pluginapi.SearchOutput
if err := c.Call(ctx, pluginapi.SearchPath("web"), pluginapi.Envelope{}, pluginapi.SearchInput{Query: "q"},
&out); err != nil {
t.Fatal(err)
}
if err := m.Deactivate(ctx, "acme.search"); err != nil {
t.Fatal(err)
}
if _, err := m.Client("acme.search"); err == nil {
t.Fatal("a removed plugin should have no client")
}
}
func TestActivateRefusesPrivateAddresses(t *testing.T) {
svc := newService(t, "1.0.0", "s3cret")
m := NewManager()
defer m.Close()
err := m.Activate(context.Background(), loaded("1.0.0", svc.URL, "s3cret"))
if err == nil || !strings.Contains(err.Error(), "not allowed") {
t.Fatalf("loopback without whitelist = %v", err)
}
}
func TestSealedSecret(t *testing.T) {
allowLoopback(t)
t.Setenv("SYSTEM_AES_KEY", strings.Repeat("k", 32))
sealed, err := utils.EncryptAESGCM("s3cret", utils.GetAESKey())
if err != nil {
t.Fatal(err)
}
svc := newService(t, "1.0.0", "s3cret")
m := NewManager()
defer m.Close()
if err := m.Activate(context.Background(), loaded("1.0.0", svc.URL, sealed)); err != nil {
t.Fatal(err)
}
}
func TestHealthChecksReportOutages(t *testing.T) {
allowLoopback(t)
svc := newService(t, "1.0.0", "s3cret")
m := NewManager()
m.interval = 10 * time.Millisecond
rep := &reports{}
m.SetReporter(rep)
defer m.Close()
if err := m.Activate(context.Background(), loaded("1.0.0", svc.URL, "s3cret")); err != nil {
t.Fatal(err)
}
svc.down.Store(true)
waitFor(t, func() bool { return len(rep.seen()) == 1 })
var pe *pluginapi.Error
if _, err := m.Client("acme.search"); !errors.As(err, &pe) || !pe.Retryable {
t.Fatalf("client while down = %v", err)
}
svc.down.Store(false)
waitFor(t, func() bool { return len(rep.seen()) == 2 })
if got := rep.seen(); got[0] || !got[1] {
t.Fatalf("reports = %v, want degraded then healthy", got)
}
if _, err := m.Client("acme.search"); err != nil {
t.Fatal(err)
}
}
func waitFor(t *testing.T, cond func() bool) {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for !cond() {
if time.Now().After(deadline) {
t.Fatal("timed out")
}
time.Sleep(5 * time.Millisecond)
}
}
+2
View File
@@ -39,6 +39,8 @@ func RegisterPluginAdminRoutes(r *gin.RouterGroup, h *handler.PluginAdminHandler
plugins.DELETE("/:id", h.UninstallPlugin)
plugins.PUT("/:id/enabled", h.SetInstalledPluginEnabled)
plugins.PUT("/:id/active-version", h.ActivatePluginVersion)
plugins.PUT("/:id/remote-url", h.SetPluginRemoteURL)
plugins.POST("/:id/secret/rotate", h.RotatePluginSecret)
plugins.GET("/:id/config", h.GetPluginSystemConfig)
plugins.PUT("/:id/config", h.UpdatePluginSystemConfig)
}
+6 -2
View File
@@ -22,8 +22,12 @@ type InstalledPlugin struct {
DesiredState string `json:"desired_state" gorm:"type:varchar(16)"`
Runtime string `json:"runtime" gorm:"type:varchar(32)"`
// GrantedPerms is the manifest permissions an administrator accepted.
GrantedPerms JSON `json:"granted_perms" gorm:"type:json"`
SystemConfig JSON `json:"system_config,omitempty" gorm:"type:json"`
GrantedPerms JSON `json:"granted_perms" gorm:"type:json"`
SystemConfig JSON `json:"system_config,omitempty" gorm:"type:json"`
// RemoteURL is where a remote plugin is served; RemoteSecret (sealed)
// signs the calls to it. Other runtimes leave both empty.
RemoteURL string `json:"remote_url,omitempty" gorm:"type:varchar(1024)"`
RemoteSecret string `json:"-" gorm:"type:text"`
CreatedBy string `json:"created_by" gorm:"type:varchar(36)"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
@@ -0,0 +1,2 @@
ALTER TABLE plugins DROP COLUMN remote_secret;
ALTER TABLE plugins DROP COLUMN remote_url;
@@ -0,0 +1,3 @@
-- Remote plugin URL and signing secret (versioned 000118).
ALTER TABLE plugins ADD COLUMN remote_url VARCHAR(1024) NOT NULL DEFAULT '';
ALTER TABLE plugins ADD COLUMN remote_secret TEXT NOT NULL DEFAULT '';
@@ -0,0 +1,2 @@
ALTER TABLE plugins DROP COLUMN IF EXISTS remote_secret;
ALTER TABLE plugins DROP COLUMN IF EXISTS remote_url;
@@ -0,0 +1,7 @@
-- Migration 000118: where a remote plugin runs and the secret its calls are
-- signed with. Only runtime.type remote plugins set them.
DO $$ BEGIN RAISE NOTICE '[Migration 000118] Adding remote plugin columns'; END $$;
ALTER TABLE plugins ADD COLUMN IF NOT EXISTS remote_url VARCHAR(1024) NOT NULL DEFAULT '';
-- Encrypted (enc:v1) HMAC secret shared with the plugin service.
ALTER TABLE plugins ADD COLUMN IF NOT EXISTS remote_secret TEXT NOT NULL DEFAULT '';
+29
View File
@@ -144,3 +144,32 @@ In Go tests, run `conformance.Run` against `httptest.NewServer(p.Handler())`.
In WeKnora: **System administration → Plugin management → Install plugin**.
Upload the `.wkp` or give its URL, review what it adds and reaches, and
install. Each workspace then enables it under **Settings → Plugins**.
## Running as a remote service
A plugin can also run as a service you deploy yourself, for instance in its
own container or on another team's cluster. Its package then carries only
`plugin.yaml` (plus schemas):
```yaml
runtime: { type: remote }
```
1. **Install.** Give the service URL when installing. WeKnora checks the
service's `/v1/manifest` against the package: same ID, version and
contributions.
2. **Keep the secret.** Installing shows a signing secret once. Start the
service with it as `WEKNORA_PLUGIN_SECRET` (and `WEKNORA_PLUGIN_ADDR`,
default `:8080`). The SDK rejects requests without a valid signature.
3. **Private hosts.** A service on a private network must be listed in
WeKnora's `SSRF_WHITELIST`.
4. **Host API.** Remote plugins get a Host API token only when
`WEKNORA_PLUGIN_HOST_API_URL` tells WeKnora its address as the service
sees it.
The plugin detail page changes the URL and rotates the secret. After a
rotation, calls fail until the service has the new secret.
Upgrade the service and the package together. While the service reports a
version other than the active package, calls are refused and the plugin
shows as degraded.