feat(plugin): dynamic form options, OAuth connections and resource limits (P2 batch 7) (#3729)

* feat(plugin): dynamic form options, OAuth connections and resource limits
* fix(plugin): show what an options list waits for as a hint, not an error
This commit is contained in:
lyingbug
2026-09-26 14:47:37 +08:00
committed by GitHub
parent cf502ba46a
commit 01222aac8e
66 changed files with 2931 additions and 9 deletions
+5
View File
@@ -826,3 +826,8 @@ CONCURRENCY_POOL_SIZE=5
# WEKNORA_PLUGIN_HOST_URL=http://plugin-host:8081
# Python 插件使用的解释器(默认 PATH 中的 python3)。
# WEKNORA_PLUGIN_PYTHON=python3
# 插件进程的内存上限(Linux),用于没有声明 runtime.resources.memory 的插件。默认不限。
# WEKNORA_PLUGIN_MEMORY_DEFAULT=1Gi
# 委托给 WeKnora 的 cgroup v2 目录(Linux,需可写)。设置后每个插件进程进入各自的子 cgroup,
# 按 runtime.resources 设置 memory.max 与 cpu.max;不设置时只限内存(rlimit),不限 CPU。
# WEKNORA_PLUGIN_CGROUP=/sys/fs/cgroup/weknora-plugins
+33
View File
@@ -161,3 +161,36 @@ export function getPluginConfig(id: string) {
export function updatePluginConfig(id: string, values: ConfigValue) {
return put<{ data: PluginConfig }>(`/api/v1/plugins/${encodeURIComponent(id)}/config`, { values })
}
/** Which configuration a plugin form edits. */
export type PluginFormScope = 'system' | 'tenant' | 'instance'
export interface PluginOptionsRequest {
name: string
field: string
scope: PluginFormScope
/** "<point>/<id>" of the instance form ("connectors/jira"). */
contribution?: string
/** The instance being edited; its stored secrets fill in redacted ones. */
instanceId?: string
/** What the form holds now, in the scope's shape. */
values: ConfigValue
query?: string
}
/** A field's choices from the plugin (x-options). */
export function pluginFormOptions(pluginId: string, req: PluginOptionsRequest) {
return post<{ data: { options: Array<{ value: string; label?: string; description?: string }> } }>(
`/api/v1/plugins/${encodeURIComponent(pluginId)}/options`, req,
)
}
/** Starts connecting an x-oauth field; the tokens stay on the server. */
export function startPluginOAuth(
pluginId: string,
req: { scope: PluginFormScope; contribution?: string; field: string },
) {
return post<{ data: { authorizeUrl: string; state: string; redirectUri: string } }>(
`/api/v1/plugins/${encodeURIComponent(pluginId)}/oauth/start`, req,
)
}
@@ -28,6 +28,7 @@
:is="resolveWidget(field.schema)"
v-else
:model-value="value[field.key]"
:path="join(field.key)"
:schema="field.schema"
:placeholder="text(field.schema, 'placeholder')"
:disabled="disabled"
@@ -0,0 +1,50 @@
import { pluginFormOptions, startPluginOAuth, type PluginFormScope } from '@/api/plugin'
import type { ConfigValue } from './schema'
import { valueAt, type SchemaFormSource } from './source'
export interface PluginFormTarget {
pluginId: string
scope: PluginFormScope
/** "<point>/<id>" for instance forms. */
contribution?: string
instanceId?: string
}
export interface PluginFormBinding {
target: () => PluginFormTarget | undefined
/** What the form holds now, in the shape the plugin's schema has. */
values: () => ConfigValue
/** A field's value by its path in the plugin's schema; defaults to a lookup in values(). */
dependency?: (path: string) => unknown
/** Maps a form path to the plugin schema's path, for forms that nest fields. */
field?: (path: string) => string
}
/** A SchemaFormSource that asks the plugin behind a form. */
export function pluginFormSource(b: PluginFormBinding): SchemaFormSource {
const need = () => {
const t = b.target()
if (!t) throw new Error('no plugin behind this form')
return t
}
const field = (path: string) => (b.field ? b.field(path) : path)
return {
dependency: path => (b.dependency ? b.dependency(path) : valueAt(b.values(), path)),
async options(path, source, query) {
const t = need()
const res = await pluginFormOptions(t.pluginId, {
name: source.name, field: field(path), scope: t.scope, contribution: t.contribution,
instanceId: t.instanceId, values: b.values(), query: query || undefined,
})
return res.data?.options ?? []
},
async startOAuth(path) {
const t = need()
const res = await startPluginOAuth(t.pluginId, {
scope: t.scope, contribution: t.contribution, field: field(path),
})
return res.data
},
}
}
+32 -1
View File
@@ -34,6 +34,32 @@ export interface ConfigSchema {
'x-group'?: string
'x-order'?: number
'x-model-types'?: string[]
'x-options'?: OptionsSource
'x-oauth'?: OAuthSpec
}
/** x-options: the field's choices come from the plugin. */
export interface OptionsSource {
/** The plugin's options endpoint. */
name: string
/** Dotted paths of fields the choices depend on; they reload when those change. */
dependsOn?: string[]
/** Ask the plugin with what the user types. */
search?: boolean
}
/** x-oauth: the field holds a connection WeKnora authorized. */
export interface OAuthSpec {
authorizeUrl: string
tokenUrl?: string
scopes?: string[]
}
/** Prefix of an x-oauth field's value (same as configschema.OAuthRefPrefix). */
export const OAUTH_REF_PREFIX = 'oauth:'
export function isOAuthRef(v: unknown): v is string {
return typeof v === 'string' && v.length > OAUTH_REF_PREFIX.length && v.startsWith(OAUTH_REF_PREFIX)
}
export type ConfigValue = Record<string, unknown>
@@ -237,6 +263,10 @@ function validateValue(
case 'string': {
if (typeof v !== 'string') return add('type')
if (schema['x-secret'] && v === REDACTED_SECRET) return
if (schema['x-oauth']) {
if (!isOAuthRef(v)) add('format')
return
}
const n = [...v].length
if (schema.minLength !== undefined && n < schema.minLength) add('min_length')
if (schema.maxLength !== undefined && n > schema.maxLength) add('max_length')
@@ -257,7 +287,8 @@ function validateValue(
if (typeof v !== 'boolean') return add('type')
break
}
const choices = choicesOf(schema)
// x-options choices are the plugin's to check.
const choices = schema['x-options'] ? [] : choicesOf(schema)
if (choices.length && !choices.some(c => looselyEqual(c.value, v))) add('enum')
}
@@ -0,0 +1,63 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import { isOAuthRef, validateConfig, type ConfigSchema } from './schema'
import { dependencyKey, isOAuthResult, optionsFailure, valueAt, type SchemaFormSource } from './source'
const schema: ConfigSchema = {
type: 'object',
required: ['account'],
properties: {
account: { type: 'string', 'x-oauth': { authorizeUrl: 'https://auth.example.com/authorize' } },
project: { type: 'string', enum: ['a'], 'x-options': { name: 'projects', dependsOn: ['account'] } },
},
}
test('x-oauth fields hold a connection reference; x-options choices are the plugin\'s to check', () => {
assert.equal(isOAuthRef('oauth:0b7c'), true)
for (const v of ['oauth:', 'token', 1, undefined]) assert.equal(isOAuthRef(v), false)
assert.deepEqual(validateConfig(schema, { account: 'oauth:1', project: 'listed-by-plugin' }), [])
assert.deepEqual(validateConfig(schema, {}), [{ path: 'account', code: 'required' }])
assert.deepEqual(validateConfig(schema, { account: 'a-token' }), [{ path: 'account', code: 'format' }])
})
test('OAuth results count only from the popup, the callback origin and the same state', () => {
const popup = {}
const start = { authorizeUrl: 'https://auth.example.com', state: 's1', redirectUri: 'https://wk.example.com/api/v1/plugin-oauth/callback' }
const data = { type: 'weknora-plugin-oauth', state: 's1', ok: true, connection: 'oauth:1' }
const ev = { data, origin: 'https://wk.example.com', source: popup }
assert.equal(isOAuthResult(ev, popup, start), true)
assert.equal(isOAuthResult({ ...ev, source: {} }, popup, start), false)
assert.equal(isOAuthResult({ ...ev, origin: 'https://evil.example.com' }, popup, start), false)
assert.equal(isOAuthResult({ ...ev, data: { ...data, state: 's2' } }, popup, start), false)
assert.equal(isOAuthResult({ ...ev, data: { ...data, type: 'other' } }, popup, start), false)
assert.equal(isOAuthResult({ ...ev, data: null }, popup, start), false)
assert.equal(isOAuthResult(ev, popup, { ...start, redirectUri: 'not a url' }), false)
})
test('choices reload only when the fields they depend on change', () => {
const values: Record<string, unknown> = { auth: { site: 'a' }, other: 1 }
const source: SchemaFormSource = {
dependency: path => valueAt(values, path),
options: async () => [],
startOAuth: async () => ({ authorizeUrl: '', state: '', redirectUri: '' }),
}
const spec = { name: 'projects', dependsOn: ['auth.site', 'missing'] }
const before = dependencyKey(source, spec)
values.other = 2
assert.equal(dependencyKey(source, spec), before)
values.auth = { site: 'b' }
assert.notEqual(dependencyKey(source, spec), before)
assert.equal(dependencyKey(source, { name: 'projects' }), '')
assert.equal(dependencyKey(undefined, spec), '')
assert.equal(valueAt({ a: [1] }, 'a.0'), undefined)
})
test('a plugin asking for more input is a hint, other failures are errors', () => {
assert.deepEqual(optionsFailure({ status: 400, message: 'connect first', error: { code: 'invalid_config', message: 'connect first' } }),
{ hint: 'connect first' })
assert.deepEqual(optionsFailure({ status: 400, message: 'bad', error: { code: 'unauthorized', message: 'token expired' } }),
{ error: 'token expired' })
assert.deepEqual(optionsFailure({ status: 502, message: 'plugin down' }), { error: 'plugin down' })
assert.deepEqual(optionsFailure(null), { error: '' })
})
@@ -0,0 +1,105 @@
import { inject, provide, type InjectionKey } from 'vue'
import type { OptionsSource } from './schema'
/** One choice a plugin offers for an x-options field. */
export interface FieldOption {
value: string
label?: string
description?: string
}
/** What starting an OAuth connection returned. */
export interface OAuthStart {
authorizeUrl: string
state: string
/** Where the provider sends the browser back; its origin posts the result. */
redirectUri: string
}
/**
* Where plugin-backed widgets (x-options, x-oauth) get their data. The page
* hosting a plugin form provides one; without it those fields fall back to
* plain inputs.
*/
export interface SchemaFormSource {
/** Current value of a field another depends on (x-options dependsOn). */
dependency(path: string): unknown
options(field: string, source: OptionsSource, query: string): Promise<FieldOption[]>
startOAuth(field: string): Promise<OAuthStart>
}
const key: InjectionKey<SchemaFormSource> = Symbol('schema-form-source')
export function provideSchemaFormSource(source: SchemaFormSource) {
provide(key, source)
}
export function useSchemaFormSource(): SchemaFormSource | undefined {
return inject(key, undefined)
}
/** The OAuth result the callback page posts to the form that opened it. */
export interface OAuthMessage {
type: 'weknora-plugin-oauth'
state: string
ok: boolean
connection?: string
error?: string
}
/**
* Whether a message event is the result of this authorization: from the
* popup, from the callback page's origin, for this state.
*/
export function isOAuthResult(
ev: { data: unknown; origin: string; source: unknown },
popup: unknown,
start: OAuthStart,
): ev is { data: OAuthMessage; origin: string; source: unknown } {
let origin = ''
try {
origin = new URL(start.redirectUri).origin
} catch {
return false
}
const d = ev.data as Partial<OAuthMessage> | null
return (
ev.source === popup &&
ev.origin === origin &&
!!d &&
d.type === 'weknora-plugin-oauth' &&
d.state === start.state
)
}
/**
* Sorts a failed options request: a plugin answering invalid_config says the
* form lacks what the list needs (no token yet), which the form shows as a
* hint; anything else is an error worth retrying.
*/
export function optionsFailure(e: unknown): { hint?: string; error?: string } {
const err = e as { message?: string; error?: { code?: string; message?: string } } | null
const message = err?.error?.message || err?.message || ''
if (err?.error?.code === 'invalid_config') return { hint: message }
return { error: message }
}
/** The values dependsOn names, keyed by path, for reload decisions. */
export function dependencyKey(source: SchemaFormSource | undefined, spec: OptionsSource | undefined): string {
if (!source || !spec?.dependsOn?.length) return ''
return JSON.stringify(spec.dependsOn.map(p => source.dependency(p) ?? null))
}
/**
* Looks a dotted path up in an object ("auth.site"); for forms that keep
* their values in one object.
*/
export function valueAt(values: Record<string, unknown> | undefined, path: string): unknown {
let v: unknown = values
for (const k of path.split('.')) {
if (typeof v !== 'object' || v === null || Array.isArray(v)) return undefined
v = (v as Record<string, unknown>)[k]
}
return v
}
@@ -0,0 +1,126 @@
<template>
<div class="oauth-widget">
<div class="oauth-widget__row">
<span class="oauth-widget__status" :class="{ 'is-connected': connected }">
<t-icon :name="connected ? 'check-circle-filled' : 'link'" />
{{ connected ? t('schemaForm.oauth.connected') : t('schemaForm.oauth.notConnected') }}
</span>
<t-button
size="small"
:variant="connected ? 'outline' : 'base'"
theme="primary"
:loading="busy"
:disabled="disabled || !source"
@click="connect"
>
{{ connected ? t('schemaForm.oauth.reconnect') : t('schemaForm.oauth.connect') }}
</t-button>
<t-button v-if="connected && !required" size="small" variant="text" :disabled="disabled || busy" @click="clear">
{{ t('schemaForm.oauth.disconnect') }}
</t-button>
</div>
<p v-if="error" class="oauth-widget__error">{{ error }}</p>
</div>
</template>
<script setup lang="ts">
import { computed, onBeforeUnmount, ref } from 'vue'
import { useI18n } from 'vue-i18n'
import { isOAuthRef } from '../schema'
import { isOAuthResult, useSchemaFormSource, type OAuthStart } from '../source'
import type { WidgetProps } from './types'
// Connects an account through the plugin's OAuth app (x-oauth). The popup
// runs the provider's consent screen; WeKnora keeps the tokens and the field
// only holds "oauth:<connection>".
const props = defineProps<WidgetProps>()
const emit = defineEmits<{ 'update:modelValue': [value: unknown] }>()
const { t } = useI18n()
const source = useSchemaFormSource()
const connected = computed(() => isOAuthRef(props.modelValue))
const busy = ref(false)
const error = ref('')
let cleanup: (() => void) | undefined
async function connect() {
if (!source) return
cleanup?.()
error.value = ''
// Open the window inside the click so popup blockers let it through.
const popup = window.open('', 'weknora-plugin-oauth', 'popup,width=560,height=720')
if (!popup) {
error.value = t('schemaForm.oauth.popupBlocked')
return
}
busy.value = true
let start: OAuthStart
try {
start = await source.startOAuth(props.path ?? '')
} catch (e) {
popup.close()
busy.value = false
error.value = (e as { message?: string })?.message || t('schemaForm.oauth.failed')
return
}
popup.location.href = start.authorizeUrl
const onMessage = (ev: MessageEvent) => {
if (!isOAuthResult(ev, popup, start)) return
done()
if (ev.data.ok && isOAuthRef(ev.data.connection)) emit('update:modelValue', ev.data.connection)
else error.value = ev.data.error || t('schemaForm.oauth.failed')
}
const closed = setInterval(() => {
if (popup.closed) done()
}, 500)
const done = () => {
window.removeEventListener('message', onMessage)
clearInterval(closed)
busy.value = false
cleanup = undefined
}
cleanup = () => {
done()
if (!popup.closed) popup.close()
}
window.addEventListener('message', onMessage)
}
function clear() {
error.value = ''
emit('update:modelValue', undefined)
}
onBeforeUnmount(() => cleanup?.())
</script>
<style lang="less" scoped>
.oauth-widget__row {
display: flex;
align-items: center;
gap: 8px;
flex-wrap: wrap;
}
.oauth-widget__status {
display: inline-flex;
align-items: center;
gap: 4px;
margin-right: auto;
color: var(--td-text-color-secondary);
font-size: var(--app-text-md);
&.is-connected {
color: var(--td-success-color);
}
}
.oauth-widget__error {
margin: 4px 0 0;
color: var(--td-error-color);
font-size: var(--app-text-sm);
line-height: 1.5;
}
</style>
@@ -0,0 +1,177 @@
<template>
<div class="options-widget">
<t-select
v-if="source"
:model-value="selected"
:options="selectOptions"
:placeholder="placeholder"
:disabled="disabled"
:clearable="!required"
:multiple="multiple"
:loading="loading"
filterable
:on-search="spec?.search ? onSearch : undefined"
@popup-visible-change="onPopup"
@update:model-value="onChange"
>
<template #empty>
<div class="options-widget__empty">{{ loading ? t('schemaForm.options.loading') : t('schemaForm.options.empty') }}</div>
</template>
</t-select>
<t-input
v-else
:model-value="Array.isArray(modelValue) ? modelValue.join(', ') : ((modelValue as string | undefined) ?? '')"
:placeholder="placeholder"
:disabled="disabled"
@update:model-value="onText"
/>
<p v-if="hint" class="options-widget__hint">{{ hint }}</p>
<p v-if="error" class="options-widget__error">
{{ error }}
<t-link theme="primary" size="small" @click="load(query)">{{ t('schemaForm.options.retry') }}</t-link>
</p>
</div>
</template>
<script setup lang="ts">
import { computed, onMounted, ref, watch } from 'vue'
import { useI18n } from 'vue-i18n'
import { dependencyKey, optionsFailure, useSchemaFormSource, type FieldOption } from '../source'
import type { WidgetProps } from './types'
// Choices the plugin lists for this field (x-options), reloaded when the
// fields they depend on change. Values the plugin no longer lists stay
// selectable so opening a form never drops a saved choice.
const props = defineProps<WidgetProps>()
const emit = defineEmits<{ 'update:modelValue': [value: unknown] }>()
const { t } = useI18n()
const source = useSchemaFormSource()
const spec = computed(() => props.schema['x-options'])
const multiple = computed(() => props.schema.type === 'array')
const options = ref<FieldOption[]>([])
const loading = ref(false)
const error = ref('')
// What the plugin needs before it can list anything ("connect an account
// first"): shown quietly, and cleared by the next load.
const hint = ref('')
const query = ref('')
let loadedKey: string | null = null
let seq = 0
const selected = computed(() => {
const v = props.modelValue
if (multiple.value) return Array.isArray(v) ? v : []
return typeof v === 'string' ? v : undefined
})
const selectOptions = computed(() => {
const out = options.value.map(o => ({ value: o.value, label: o.label || o.value, title: o.description }))
const known = new Set(out.map(o => o.value))
const current = Array.isArray(selected.value) ? selected.value : selected.value ? [selected.value] : []
for (const v of current) {
if (typeof v === 'string' && !known.has(v)) out.push({ value: v, label: v, title: undefined })
}
return out
})
async function load(q = '') {
if (!source || !spec.value) return
const mine = ++seq
loading.value = true
error.value = ''
hint.value = ''
try {
const got = await source.options(props.path ?? '', spec.value, q)
if (mine === seq) options.value = got
} catch (e) {
if (mine === seq) {
options.value = []
const failure = optionsFailure(e)
if (failure.hint !== undefined) hint.value = failure.hint
else error.value = failure.error || t('schemaForm.options.failed')
}
} finally {
if (mine === seq) loading.value = false
}
}
const depKey = computed(() => dependencyKey(source, spec.value))
onMounted(() => {
loadedKey = depKey.value
void load()
})
// Typing into a field the choices depend on reloads them once it settles.
let timer: ReturnType<typeof setTimeout> | undefined
watch(depKey, key => {
clearTimeout(timer)
timer = setTimeout(() => {
if (key === loadedKey) return
loadedKey = key
void load(query.value)
}, 400)
})
function onPopup(visible: boolean) {
if (visible && error.value && !loading.value) void load(query.value)
}
let searchTimer: ReturnType<typeof setTimeout> | undefined
function onSearch(q: string) {
query.value = q
clearTimeout(searchTimer)
searchTimer = setTimeout(() => void load(q), 300)
}
function onChange(v: unknown) {
if (multiple.value) {
emit('update:modelValue', Array.isArray(v) && v.length ? v : undefined)
return
}
emit('update:modelValue', v === '' || v === null ? undefined : v)
}
function onText(v: string) {
if (!multiple.value) {
emit('update:modelValue', v || undefined)
return
}
const list = v.split(',').map(s => s.trim()).filter(Boolean)
emit('update:modelValue', list.length ? list : undefined)
}
</script>
<style lang="less" scoped>
.options-widget {
display: flex;
flex-direction: column;
gap: 4px;
}
.options-widget__empty {
padding: 8px 12px;
color: var(--td-text-color-placeholder);
font-size: var(--app-text-sm);
}
.options-widget__hint {
margin: 0;
color: var(--td-text-color-placeholder);
font-size: var(--app-text-sm);
line-height: 1.5;
}
.options-widget__error {
display: flex;
align-items: center;
gap: 8px;
margin: 0;
color: var(--td-error-color);
font-size: var(--app-text-sm);
line-height: 1.5;
}
</style>
@@ -3,6 +3,8 @@ import type { Component } from 'vue'
import type { ConfigSchema } from '../schema'
import HeadersWidget from './HeadersWidget.vue'
import NumberWidget from './NumberWidget.vue'
import OAuthWidget from './OAuthWidget.vue'
import OptionsWidget from './OptionsWidget.vue'
import PasswordWidget from './PasswordWidget.vue'
import SelectWidget from './SelectWidget.vue'
import SwitchWidget from './SwitchWidget.vue'
@@ -22,6 +24,10 @@ const widgets = new Map<string, Component>([
['switch', SwitchWidget],
// "Name: Value" lines edited as rows (RSS auth headers).
['headers', HeadersWidget],
// Choices loaded from the plugin (x-options).
['options', OptionsWidget],
// A connect button for an OAuth connection (x-oauth).
['oauth', OAuthWidget],
])
export function registerWidget(name: string, component: Component) {
@@ -32,7 +38,9 @@ export function registerWidget(name: string, component: Component) {
export function widgetName(schema: ConfigSchema): string {
const explicit = schema['x-widget']
if (explicit && widgets.has(explicit)) return explicit
if (schema['x-oauth']) return 'oauth'
if (schema['x-secret']) return 'password'
if (schema['x-options']) return 'options'
if (schema.oneOf?.length || schema.enum?.length) return 'select'
if (schema.type === 'boolean') return 'switch'
if (schema.type === 'number' || schema.type === 'integer') return 'number'
@@ -3,6 +3,8 @@ import type { ConfigSchema } from '../schema'
/** Props every widget receives. Widgets emit update:modelValue. */
export interface WidgetProps {
modelValue: unknown
/** Dotted path of the field in its form. */
path?: string
schema: ConfigSchema
placeholder?: string
disabled?: boolean
+15
View File
@@ -2916,6 +2916,21 @@ export default {
minimum: 'Value is too small',
maximum: 'Value is too large',
format: 'Enter a full address such as https://example.com'
},
options: {
loading: 'Loading…',
empty: 'The plugin offers no choices',
retry: 'Retry',
failed: 'Could not load the choices'
},
oauth: {
connected: 'Connected',
notConnected: 'Not connected',
connect: 'Connect',
reconnect: 'Reconnect',
disconnect: 'Disconnect',
popupBlocked: 'The browser blocked the sign-in window. Allow pop-ups and try again.',
failed: 'Authorization failed'
}
},
common: {
+15
View File
@@ -2916,6 +2916,21 @@ export default {
minimum: '値が小さすぎます',
maximum: '値が大きすぎます',
format: 'https://example.com のような完全なアドレスを入力してください'
},
options: {
loading: '読み込み中…',
empty: '選択肢がありません',
retry: '再試行',
failed: '選択肢を読み込めませんでした'
},
oauth: {
connected: '接続済み',
notConnected: '未接続',
connect: 'アカウントを接続',
reconnect: '再接続',
disconnect: '切断',
popupBlocked: 'ブラウザが認可ウィンドウをブロックしました。ポップアップを許可して再試行してください。',
failed: '認可に失敗しました'
}
},
common: {
+15
View File
@@ -5434,6 +5434,21 @@ export default {
minimum: '값이 너무 작습니다',
maximum: '값이 너무 큽니다',
format: 'https://example.com 과 같은 전체 주소를 입력하세요'
},
options: {
loading: '불러오는 중…',
empty: '선택할 수 있는 항목이 없습니다',
retry: '다시 시도',
failed: '선택 항목을 불러오지 못했습니다'
},
oauth: {
connected: '연결됨',
notConnected: '연결되지 않음',
connect: '계정 연결',
reconnect: '다시 연결',
disconnect: '연결 해제',
popupBlocked: '브라우저가 인증 창을 차단했습니다. 팝업을 허용한 뒤 다시 시도하세요.',
failed: '인증에 실패했습니다'
}
},
common: {
+15
View File
@@ -5434,6 +5434,21 @@ export default {
minimum: 'Слишком маленькое значение',
maximum: 'Слишком большое значение',
format: 'Введите полный адрес, например https://example.com'
},
options: {
loading: 'Загрузка…',
empty: 'Плагин не предлагает вариантов',
retry: 'Повторить',
failed: 'Не удалось загрузить варианты'
},
oauth: {
connected: 'Подключено',
notConnected: 'Не подключено',
connect: 'Подключить',
reconnect: 'Переподключить',
disconnect: 'Отключить',
popupBlocked: 'Браузер заблокировал окно авторизации. Разрешите всплывающие окна и повторите попытку.',
failed: 'Не удалось авторизоваться'
}
},
common: {
+15
View File
@@ -5436,6 +5436,21 @@ export default {
minimum: '数值过小',
maximum: '数值过大',
format: '请输入完整的地址,如 https://example.com'
},
options: {
loading: '加载中…',
empty: '插件没有可选项',
retry: '重试',
failed: '加载可选项失败'
},
oauth: {
connected: '已连接',
notConnected: '未连接',
connect: '连接账号',
reconnect: '重新连接',
disconnect: '断开',
popupBlocked: '浏览器拦截了授权窗口,请允许弹出窗口后重试',
failed: '授权失败'
}
},
common: {
@@ -11,6 +11,7 @@ import { createRenderer, h, nextTick, reactive, ref } from 'vue'
const require = createRequire(import.meta.url)
// The dialog validates credentials with the real schema helpers.
const schemaModule = require('../../../components/schema-form/schema.ts')
const sourceModule = require('../../../components/schema-form/source.ts')
const filename = fileURLToPath(new URL('./DataSourceEditorDialog.vue', import.meta.url))
const { descriptor } = parse(readFileSync(filename, 'utf8'), { filename })
const script = compileScript(descriptor, { id: 'datasource-editor-test' }).content
@@ -46,7 +47,10 @@ async function fixture({
access_token: { type: 'string', title: 'Personal access token', 'x-secret': true, 'x-order': 2 },
},
},
}, { type: 'yuque', config_schema: { type: 'object', properties: {} } }]
}, { type: 'yuque', config_schema: { type: 'object', properties: {} } }, {
type: 'acme.jira/jira', plugin_id: 'acme.jira',
config_schema: { type: 'object', properties: { site: { type: 'string' } } },
}]
},
async validateCredentials(type: string, credentials: Record<string, string>) {
calls.push({ method: 'validateCredentials', args: [type, { ...credentials }] })
@@ -76,6 +80,7 @@ async function fixture({
},
})
const exports: any = {}
let formBinding: any
runInNewContext(compiled, {
exports,
require(name: string) {
@@ -92,6 +97,12 @@ async function fixture({
if (name === 'tdesign-vue-next') return { MessagePlugin: { warning() {}, success() {}, error() {} } }
if (name === '@/api/datasource') return api
if (name === '@/components/schema-form/schema') return schemaModule
if (name === '@/components/schema-form/source') {
return { valueAt: sourceModule.valueAt, provideSchemaFormSource() {} }
}
if (name === '@/components/schema-form/pluginSource') {
return { pluginFormSource: (b: unknown) => { formBinding = b; return {} } }
}
if (name === '@/components/schema-form/useSchemaText') {
return { useSchemaText: () => (schema: { title?: string }) => schema.title ?? '' }
}
@@ -117,7 +128,9 @@ async function fixture({
vm.form.config.credentials = { base_url: 'https://gitlab.example.com', access_token: token }
await nextTick()
}
return { vm, calls, replace, storedToken: () => storedToken, close: () => app.unmount() }
return {
vm, calls, replace, formBinding, storedToken: () => storedToken, close: () => app.unmount(),
}
}
test('rotated GitLab credentials are tested without updating the saved data source', async () => {
@@ -270,3 +283,21 @@ test('Cloud hierarchy limitation stays visible after an empty space expansion',
assert.equal(f.vm.visibleTree.some((row: any) => row.noticeAfter), true)
} finally { f.close() }
})
test('plugin connector forms ask the plugin behind the connector type', async () => {
const f = await fixture({ type: 'acme.jira/jira', settings: { project: 'p1' } })
try {
await new Promise((resolve) => setTimeout(resolve, 0))
assert.deepEqual(JSON.parse(JSON.stringify(f.formBinding.target())), {
pluginId: 'acme.jira', scope: 'instance', contribution: 'connectors/jira', instanceId: 'source-one',
})
f.vm.form.config.credentials = { site: 'acme' }
assert.equal(f.formBinding.dependency('site'), 'acme')
assert.equal(f.formBinding.dependency('project'), 'p1')
assert.deepEqual(JSON.parse(JSON.stringify(f.formBinding.values())), {
credentials: { site: 'acme' }, settings: { project: 'p1' }, resourceIds: [],
})
f.vm.form.type = 'gitlab'
assert.equal(f.formBinding.target(), undefined)
} finally { f.close() }
})
@@ -26,6 +26,8 @@ import {
type ConfigSchema,
type FieldError,
} from '@/components/schema-form/schema'
import { pluginFormSource } from '@/components/schema-form/pluginSource'
import { provideSchemaFormSource, valueAt } from '@/components/schema-form/source'
import { useSchemaText } from '@/components/schema-form/useSchemaText'
import SettingDrawer from '@/components/settings/SettingDrawer.vue'
import DataSourceTypeIcon from './DataSourceTypeIcon.vue'
@@ -540,6 +542,27 @@ const credentialSchema = computed<ConfigSchema>(() => currentDef.value?.config_s
// Settings form of connectors without a built-in settings UI (plugins).
const settingsSchema = computed<ConfigSchema | null>(() => currentDef.value?.settings_schema ?? null)
const settingsErrors = ref<FieldError[]>([])
// Plugin connector fields with x-options / x-oauth ask the plugin, with the
// instance the form holds (the stored credentials fill in what edit mode hides).
provideSchemaFormSource(pluginFormSource({
target: () => {
const def = currentDef.value
if (!def?.plugin_id || !def.type.startsWith(`${def.plugin_id}/`)) return undefined
return {
pluginId: def.plugin_id,
scope: 'instance',
contribution: `connectors/${def.type.slice(def.plugin_id.length + 1)}`,
instanceId: props.dataSource?.id,
}
},
values: () => ({
credentials: form.value.config.credentials,
settings: form.value.config.settings,
resourceIds: form.value.config.resource_ids,
}),
dependency: path => valueAt(form.value.config.credentials, path) ?? valueAt(form.value.config.settings, path),
}))
const labelI18n = computed(() => ({ t: (k: string) => t(k), te: (k: string) => te(k), locale: locale.value }))
const labelOf = (def: ConnectorMeta) => connectorName(def, labelI18n.value)
@@ -123,6 +123,8 @@ import {
type TenantPlugin,
} from '@/api/plugin'
import SchemaForm from '@/components/schema-form/SchemaForm.vue'
import { pluginFormSource } from '@/components/schema-form/pluginSource'
import { provideSchemaFormSource } from '@/components/schema-form/source'
import { validateConfig, type ConfigSchema, type ConfigValue, type FieldError } from '@/components/schema-form/schema'
import SettingDrawer from '@/components/settings/SettingDrawer.vue'
import { useAuthStore } from '@/stores/auth'
@@ -210,6 +212,12 @@ const configErrors = ref<FieldError[]>([])
const configLoading = ref(false)
const configSaving = ref(false)
const configWebhooks = ref<PluginWebhook[]>([])
// x-options / x-oauth fields of the workspace configuration ask the plugin.
provideSchemaFormSource(pluginFormSource({
target: () => (configPlugin.value ? { pluginId: configPlugin.value.manifest.id, scope: 'tenant' } : undefined),
values: () => configValues.value,
}))
const origin = window.location.origin
async function openConfig(p: TenantPlugin) {
@@ -271,6 +271,8 @@ import {
} from '@/api/web-search-provider'
import SettingDrawer from '@/components/settings/SettingDrawer.vue'
import SchemaForm from '@/components/schema-form/SchemaForm.vue'
import { pluginFormSource } from '@/components/schema-form/pluginSource'
import { provideSchemaFormSource } from '@/components/schema-form/source'
import {
inGroup,
isPlainObject,
@@ -355,6 +357,28 @@ const EMPTY_SCHEMA: ConfigSchema = { type: 'object', properties: {} }
const parametersSchema = computed<ConfigSchema>(() => selectedProviderType.value?.config_schema ?? EMPTY_SCHEMA)
// Plugin providers' x-options / x-oauth fields ask the plugin. Its schema is
// flat: api_key plus fields the form keeps under extra_config.
const EXTRA = 'extra_config.'
function pluginInstanceValues(): ConfigValue {
const { extra_config: extra, ...rest } = providerForm.value.parameters
return { ...rest, ...(extra && typeof extra === 'object' ? (extra as ConfigValue) : {}) }
}
provideSchemaFormSource(pluginFormSource({
target: () => {
const type = selectedProviderType.value
if (!type?.plugin_id || !type.id.startsWith(`${type.plugin_id}/`)) return undefined
return {
pluginId: type.plugin_id,
scope: 'instance',
contribution: `webSearch/${type.id.slice(type.plugin_id.length + 1)}`,
instanceId: editingProvider.value?.id,
}
},
values: pluginInstanceValues,
field: path => (path.startsWith(EXTRA) ? path.slice(EXTRA.length) : path),
}))
const hasGroup = (group: string) => orderedFields(parametersSchema.value).some(f => inGroup(f.schema, group))
// Validation for the parameters. In edit mode secrets are committed through
@@ -142,6 +142,8 @@ import { MessagePlugin } from 'tdesign-vue-next'
import SettingDrawer from '@/components/settings/SettingDrawer.vue'
import SchemaForm from '@/components/schema-form/SchemaForm.vue'
import { pluginFormSource } from '@/components/schema-form/pluginSource'
import { provideSchemaFormSource } from '@/components/schema-form/source'
import { validateConfig, type ConfigSchema, type ConfigValue, type FieldError } from '@/components/schema-form/schema'
import { getPlugin, type PluginInstance } from '@/api/plugin'
import {
@@ -183,6 +185,12 @@ const instances = ref<PluginInstance[]>([])
const instanceError = ref('')
const systemSchema = ref<ConfigSchema | null>(null)
const configValues = ref<ConfigValue>({})
// x-options / x-oauth fields of the platform configuration ask the plugin.
provideSchemaFormSource(pluginFormSource({
target: () => (props.plugin ? { pluginId: props.plugin.id, scope: 'system' } : undefined),
values: () => configValues.value,
}))
const configErrors = ref<FieldError[]>([])
const saving = ref(false)
const activating = ref('')
@@ -0,0 +1,49 @@
package repository
import (
"context"
"errors"
"time"
"gorm.io/gorm"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
)
type pluginOAuthRepository struct{ db *gorm.DB }
// NewPluginOAuthRepository creates the plugin OAuth connection repository.
func NewPluginOAuthRepository(db *gorm.DB) interfaces.PluginOAuthRepository {
return &pluginOAuthRepository{db: db}
}
func (r *pluginOAuthRepository) Create(ctx context.Context, c *types.PluginOAuthConnection) error {
return r.db.WithContext(ctx).Create(c).Error
}
func (r *pluginOAuthRepository) Get(
ctx context.Context, pluginID string, tenantID uint64, id string,
) (*types.PluginOAuthConnection, error) {
var c types.PluginOAuthConnection
err := r.db.WithContext(ctx).
Where("id = ? AND plugin_id = ? AND tenant_id = ?", id, pluginID, tenantID).
First(&c).
Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
if err != nil {
return nil, err
}
return &c, nil
}
func (r *pluginOAuthRepository) UpdateToken(
ctx context.Context, c *types.PluginOAuthConnection, prev time.Time,
) (bool, error) {
res := r.db.WithContext(ctx).Model(&types.PluginOAuthConnection{}).
Where("id = ? AND updated_at = ?", c.ID, prev).
Updates(map[string]any{"token": c.Token, "expires_at": c.ExpiresAt, "updated_at": c.UpdatedAt})
return res.RowsAffected == 1, res.Error
}
+4
View File
@@ -76,6 +76,7 @@ import (
plugindriver "github.com/Tencent/WeKnora/internal/plugin/driver"
pluginevents "github.com/Tencent/WeKnora/internal/plugin/events"
pluginmanifest "github.com/Tencent/WeKnora/internal/plugin/manifest"
pluginoauth "github.com/Tencent/WeKnora/internal/plugin/oauth"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
pluginremote "github.com/Tencent/WeKnora/internal/plugin/remote"
@@ -627,6 +628,9 @@ func BuildContainer(container *dig.Container) *dig.Container {
must(container.Provide(handler.NewPluginUIHandler))
must(container.Provide(pluginwebhook.NewTokensFromEnv))
must(container.Provide(handler.NewPluginWebhookHandler))
must(container.Provide(repository.NewPluginOAuthRepository))
must(container.Provide(pluginoauth.NewService))
must(container.Provide(handler.NewPluginFormsHandler))
logger.Debugf(ctx, "[Container] HTTP handlers registered")
// Wire the chat package's local image resolver so multimodal chat can read
+3 -1
View File
@@ -23,6 +23,7 @@ import (
"github.com/Tencent/WeKnora/internal/plugin/hostpool"
"github.com/Tencent/WeKnora/internal/plugin/install"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
pluginoauth "github.com/Tencent/WeKnora/internal/plugin/oauth"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
"github.com/Tencent/WeKnora/internal/plugin/remote"
@@ -73,9 +74,10 @@ func (a pluginActivators) list() []reconcile.Activator {
// (remote, on a plugin host) only through WEKNORA_PLUGIN_HOST_API_URL.
func newPluginHostAPI(
cfg *config.Config, iv *activate.Invoker, repo interfaces.PluginKVRepository, cleaner interfaces.ResourceCleaner,
hooks *webhook.Tokens,
hooks *webhook.Tokens, oauth *pluginoauth.Service,
) *hostapi.Handler {
iv.SetWebhooks(hooks, webhook.PublicBase())
iv.SetOAuth(oauth)
issuer := hostapi.NewIssuerFromEnv()
// Plugins on this node always use loopback; the public address is for
// plugins elsewhere (remote, on a plugin host), which reach this node
@@ -73,7 +73,7 @@ var versionedSQLiteColumns = map[string][]string{
}, // 000028
}
const expectedSQLiteMigrationVersion = 37
const expectedSQLiteMigrationVersion = 38
func TestSQLiteMigrationsCreateVersionedSchema(t *testing.T) {
repoRoot := sqliteRepoRoot(t)
+366
View File
@@ -0,0 +1,366 @@
package handler
import (
"encoding/json"
stderrors "errors"
"html/template"
"net/http"
"net/url"
"os"
"strings"
"github.com/gin-gonic/gin"
"github.com/Tencent/WeKnora/internal/errors"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/plugin/activate"
"github.com/Tencent/WeKnora/internal/plugin/configschema"
"github.com/Tencent/WeKnora/internal/plugin/install"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
pluginoauth "github.com/Tencent/WeKnora/internal/plugin/oauth"
"github.com/Tencent/WeKnora/internal/plugin/registry"
"github.com/Tencent/WeKnora/internal/plugin/tenancy"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// PluginFormsHandler serves what plugin forms need from the plugin while
// they are being filled: dynamic choices (x-options) and OAuth connections
// (x-oauth).
type PluginFormsHandler struct {
registry *registry.Registry
tenancy *tenancy.Service
invoker *activate.Invoker
plugins interfaces.PluginRepository
dataSources interfaces.DataSourceRepository
webSearch interfaces.WebSearchProviderRepository
oauth *pluginoauth.Service
}
// NewPluginFormsHandler creates the handler.
func NewPluginFormsHandler(
reg *registry.Registry, t *tenancy.Service, iv *activate.Invoker, plugins interfaces.PluginRepository,
dataSources interfaces.DataSourceRepository, webSearch interfaces.WebSearchProviderRepository,
oauth *pluginoauth.Service,
) *PluginFormsHandler {
return &PluginFormsHandler{
registry: reg, tenancy: t, invoker: iv, plugins: plugins, dataSources: dataSources, webSearch: webSearch,
oauth: oauth,
}
}
// PluginOptionsRequest asks a plugin for a field's choices.
type PluginOptionsRequest struct {
Name string `json:"name" binding:"required"`
Field string `json:"field" binding:"required"`
// Scope is system, tenant or instance.
Scope string `json:"scope" binding:"required"`
// Contribution is "<point>/<id>" of the instance ("connectors/jira").
Contribution string `json:"contribution"`
// InstanceID is the instance being edited; its stored secrets fill in
// the fields the form did not change.
InstanceID string `json:"instanceId"`
// Values are what the form holds now, in the scope's shape.
Values map[string]any `json:"values"`
Query string `json:"query"`
}
// Options godoc
// @Summary 插件表单的动态选项
// @Description 用表单当前的值(未修改的密钥用已存的值补齐)向插件取某个字段的可选项(x-options)
// @Tags Plugins
// @Accept json
// @Produce json
// @Param id path string true "插件 ID"
// @Param request body PluginOptionsRequest true "请求"
// @Success 200 {object} map[string]interface{}
// @Security Bearer
// @Router /plugins/{id}/options [post]
func (h *PluginFormsHandler) Options(c *gin.Context) {
limitJSONBody(c, 1<<20)
var req PluginOptionsRequest
if err := c.ShouldBindJSON(&req); err != nil {
_ = c.Error(errors.NewBadRequestError("name, field and scope are required"))
return
}
ctx := c.Request.Context()
m, ok := h.codePlugin(c, c.Param("id"), req.Scope)
if !ok {
return
}
tenantID := c.GetUint64(types.TenantIDContextKey.String())
values := req.Values
if values == nil {
values = map[string]any{}
}
var override activate.ConfigOverride
var instance map[string]any
switch req.Scope {
case pluginapi.OptionsScopeSystem:
stored, _, err := install.OpenSystemConfig(ctx, h.plugins, m)
if err != nil {
h.fail(c, err)
return
}
override.System = mergeForm(m.Config.SystemSchema, stored, values)
case pluginapi.OptionsScopeTenant:
stored, _, err := h.tenancy.OpenConfig(ctx, tenantID, m.ID)
if err != nil {
h.fail(c, err)
return
}
override.Tenant = mergeForm(m.Config.TenantSchema, stored, values)
case pluginapi.OptionsScopeInstance:
var err error
if instance, err = h.instanceValues(c, tenantID, req, values); err != nil {
h.fail(c, err)
return
}
default:
_ = c.Error(errors.NewBadRequestError("scope must be system, tenant or instance"))
return
}
in := pluginapi.OptionsInput{Field: req.Field, Scope: req.Scope, Contribution: req.Contribution, Query: req.Query}
var out pluginapi.OptionsOutput
err := h.invoker.CallOverriding(ctx, m, pluginapi.OptionsPath(req.Name), instance, override, in, &out)
if err != nil {
h.fail(c, err)
return
}
if out.Options == nil {
out.Options = []pluginapi.Option{}
}
c.JSON(http.StatusOK, gin.H{"success": true, "data": out})
}
// codePlugin finds a plugin with code the caller may ask for this scope.
func (h *PluginFormsHandler) codePlugin(c *gin.Context, id, scope string) (*manifest.Manifest, bool) {
ctx := c.Request.Context()
m, ok := h.registry.Plugin(id)
if !ok || (m.Runtime.Type != manifest.RuntimeHost && m.Runtime.Type != manifest.RuntimeRemote) {
_ = c.Error(errors.NewNotFoundError("no such plugin with code"))
return nil, false
}
if scope == pluginapi.OptionsScopeSystem {
if !types.IsSystemAdminFromContext(ctx) {
c.AbortWithStatusJSON(http.StatusForbidden, gin.H{"error": "Forbidden: system administrators only"})
return nil, false
}
return m, true
}
tenantID := c.GetUint64(types.TenantIDContextKey.String())
if on, err := h.tenancy.PluginEnabled(ctx, tenantID, id); err != nil || !on {
_ = c.Error(errors.NewNotFoundError("the plugin is not enabled in this workspace"))
return nil, false
}
return m, true
}
// mergeForm fills the fields a form left redacted ("***") or empty with the
// stored values.
func mergeForm(rawSchema json.RawMessage, stored, values map[string]any) map[string]any {
if len(rawSchema) == 0 {
return values
}
schema, err := configschema.Parse(rawSchema)
if err != nil {
return values
}
return configschema.Merge(schema, stored, values)
}
// instanceValues is the instance the form holds, with the stored instance's
// secrets filled in where the form has none (forms never receive secrets).
func (h *PluginFormsHandler) instanceValues(
c *gin.Context, tenantID uint64, req PluginOptionsRequest, values map[string]any,
) (map[string]any, error) {
point, _, _ := strings.Cut(req.Contribution, "/")
switch manifest.Point(point) {
case manifest.PointConnectors:
creds, _ := values["credentials"].(map[string]any)
if creds == nil {
creds = map[string]any{}
}
if req.InstanceID != "" {
ds, err := h.dataSources.FindByID(c.Request.Context(), req.InstanceID)
if err != nil || ds == nil || ds.TenantID != tenantID {
return nil, errInstanceNotFound
}
cfg, err := ds.ParseConfig()
if err != nil {
return nil, err
}
for k, v := range cfg.Credentials {
if s, _ := creds[k].(string); s == "" || s == types.RedactedSecretPlaceholder {
creds[k] = v
}
}
}
out := map[string]any{
"credentials": creds,
"settings": values["settings"],
"resourceIds": values["resourceIds"],
}
if out["settings"] == nil {
out["settings"] = map[string]any{}
}
return out, nil
case manifest.PointWebSearch:
out := map[string]any{}
for k, v := range values {
out[k] = v
}
if key, _ := out["api_key"].(string); req.InstanceID != "" &&
(key == "" || key == types.RedactedSecretPlaceholder) {
p, err := h.webSearch.GetByID(c.Request.Context(), tenantID, req.InstanceID)
if err != nil || p == nil {
return nil, errInstanceNotFound
}
out["api_key"] = p.Parameters.APIKey
}
return out, nil
default:
return values, nil
}
}
var errInstanceNotFound = stderrors.New("instance not found")
// fail maps a plugin form failure: the plugin's own field problems go back
// to the form as they are.
func (h *PluginFormsHandler) fail(c *gin.Context, err error) {
if stderrors.Is(err, errInstanceNotFound) {
_ = c.Error(errors.NewNotFoundError("instance not found"))
return
}
if pe, ok := pluginapi.AsError(err); ok {
switch pe.Code {
case pluginapi.CodeInvalidConfig, pluginapi.CodeUnauthorized, pluginapi.CodeBadRequest:
body := gin.H{"success": false, "error": gin.H{"code": string(pe.Code), "message": pe.Message}}
if pe.Details != nil && len(pe.Details.Fields) > 0 {
body["error"].(gin.H)["fields"] = pe.Details.Fields
}
c.JSON(http.StatusBadRequest, body)
return
case pluginapi.CodeNotFound:
_ = c.Error(errors.NewNotFoundError(pe.Message))
return
}
}
logger.Warnf(c.Request.Context(), "[plugin] form request to %s failed: %v", c.Param("id"), err)
c.JSON(http.StatusBadGateway, gin.H{"success": false, "error": gin.H{"message": err.Error()}})
}
// PluginOAuthStartRequest starts connecting an x-oauth field.
type PluginOAuthStartRequest struct {
Scope string `json:"scope" binding:"required"`
Contribution string `json:"contribution"`
Field string `json:"field" binding:"required"`
}
// OAuthStart godoc
// @Summary 开始插件字段的 OAuth 授权
// @Description 为 x-oauth 字段生成授权地址;授权完成后回调页把连接 ID 交回表单,令牌只保存在服务端
// @Tags Plugins
// @Accept json
// @Produce json
// @Param id path string true "插件 ID"
// @Param request body PluginOAuthStartRequest true "字段"
// @Success 200 {object} map[string]interface{}
// @Security Bearer
// @Router /plugins/{id}/oauth/start [post]
func (h *PluginFormsHandler) OAuthStart(c *gin.Context) {
var req PluginOAuthStartRequest
if err := c.ShouldBindJSON(&req); err != nil {
_ = c.Error(errors.NewBadRequestError("scope and field are required"))
return
}
m, ok := h.codePlugin(c, c.Param("id"), req.Scope)
if !ok {
return
}
target := pluginoauth.Target{
PluginID: m.ID, Scope: req.Scope, Contribution: req.Contribution, Field: req.Field,
}
if req.Scope != pluginapi.OptionsScopeSystem {
target.TenantID = c.GetUint64(types.TenantIDContextKey.String())
}
userID, _ := types.UserIDFromContext(c.Request.Context())
origin, base := appOrigin(c)
url, state, err := h.oauth.Start(c.Request.Context(), target, userID, origin, base)
if err != nil {
_ = c.Error(errors.NewBadRequestError(err.Error()))
return
}
c.JSON(http.StatusOK, gin.H{"success": true, "data": gin.H{
"authorizeUrl": url, "state": state, "redirectUri": base + pluginoauth.CallbackPath,
}})
}
// appOrigin is the app's origin (where the callback page reports back) and
// the base URL authorization servers send the browser to: APP_EXTERNAL_URL
// when set, else the origin the browser called from.
func appOrigin(c *gin.Context) (origin, base string) {
if ext := strings.TrimSuffix(strings.TrimSpace(os.Getenv("APP_EXTERNAL_URL")), "/"); ext != "" {
if u, err := url.Parse(ext); err == nil && u.Host != "" {
return u.Scheme + "://" + u.Host, ext
}
}
if o := c.GetHeader("Origin"); strings.HasPrefix(o, "http://") || strings.HasPrefix(o, "https://") {
return o, o
}
scheme := "http"
if c.Request.TLS != nil || strings.EqualFold(c.GetHeader("X-Forwarded-Proto"), "https") {
scheme = "https"
}
host := c.Request.Host
if fh := c.GetHeader("X-Forwarded-Host"); fh != "" {
host = fh
}
o := scheme + "://" + host
return o, o
}
var oauthDoneTemplate = template.Must(template.New("oauth").Parse(`<!doctype html>
<html><head><meta charset="utf-8"><title>WeKnora</title></head>
<body style="font:14px system-ui,sans-serif;padding:24px">
<p id="msg">{{.Message}}</p>
<script>
(function () {
var msg = {
type: 'weknora-plugin-oauth', state: {{.State}}, ok: {{.OK}},
connection: {{.Connection}}, error: {{.Error}}
};
if (window.opener && {{.Origin}}) {
window.opener.postMessage(msg, {{.Origin}});
window.close();
}
})();
</script>
</body></html>`))
// OAuthCallback is where authorization servers send the browser back. It
// needs no login (the state names the pending authorization) and answers
// with a page that hands the result to the form that opened it.
func (h *PluginFormsHandler) OAuthCallback(c *gin.Context) {
res := h.oauth.Complete(c.Request.Context(), c.Query("state"), c.Query("code"), c.Query("error"))
data := map[string]any{
"State": res.State, "Origin": res.Origin, "OK": res.Err == nil, "Connection": "", "Error": "",
"Message": "Connected. You can close this window.",
}
if res.Err != nil {
data["Error"], data["Message"] = res.Err.Error(), res.Err.Error()
} else {
data["Connection"] = configschema.OAuthRefPrefix + res.ConnectionID
}
c.Header("Content-Type", "text/html; charset=utf-8")
c.Header("Cache-Control", "no-store")
c.Header("Referrer-Policy", "no-referrer")
status := http.StatusOK
if res.Err != nil {
status = http.StatusBadRequest
}
c.Status(status)
_ = oauthDoneTemplate.Execute(c.Writer, data)
}
+147
View File
@@ -0,0 +1,147 @@
package handler
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/gin-gonic/gin"
"github.com/Tencent/WeKnora/internal/middleware"
"github.com/Tencent/WeKnora/internal/plugin/activate"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
pluginoauth "github.com/Tencent/WeKnora/internal/plugin/oauth"
"github.com/Tencent/WeKnora/internal/plugin/plugintest"
"github.com/Tencent/WeKnora/internal/plugin/registry"
"github.com/Tencent/WeKnora/internal/plugin/tenancy"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/pluginsdk"
"github.com/Tencent/WeKnora/pluginsdk/client"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
func formsEngine(t *testing.T) (*gin.Engine, *plugintest.MemTenantSettings) {
t.Helper()
gin.SetMode(gin.TestMode)
tenantSchema, _ := json.Marshal(map[string]any{"type": "object", "properties": map[string]any{
"site": map[string]any{"type": "string"},
"token": map[string]any{"type": "string", "x-secret": true},
"project": map[string]any{
"type": "string",
"x-options": map[string]any{"name": "projects", "dependsOn": []string{"site", "token"}},
},
}})
m := &manifest.Manifest{
SchemaVersion: manifest.SchemaVersion, ID: "acme.jira", Version: "1.0.0", APIVersion: pluginapi.APIVersion,
Name: manifest.Text("Jira", nil), Publisher: manifest.Publisher{ID: "acme"},
Runtime: manifest.Runtime{Type: manifest.RuntimeHost, Kind: "binary", Entry: "bin/x"},
Contributes: manifest.Contributions{manifest.PointWebSearch: {{ID: "x", Name: manifest.Text("X", nil)}}},
Config: manifest.ConfigSchemas{TenantSchema: tenantSchema},
}
reg := registry.New()
if err := reg.Register(m); err != nil {
t.Fatal(err)
}
settings := &plugintest.MemTenantSettings{}
ten := tenancy.NewService(reg, settings)
ctx := context.Background()
_ = settings.Upsert(ctx, &types.PluginTenantSetting{TenantID: 7, PluginID: "acme.jira", Enabled: true})
stored := map[string]any{"site": "old", "token": "stored-secret"}
if _, err := ten.SetConfig(ctx, 7, "acme.jira", stored, "u"); err != nil {
t.Fatal(err)
}
p := pluginsdk.New(pluginsdk.Info{ID: "acme.jira", Version: "1.0.0"})
p.Options(
"projects",
func(_ context.Context, call *pluginsdk.Call, in pluginapi.OptionsInput) ([]pluginapi.Option, error) {
if call.Config.Tenant["token"] != "stored-secret" {
return nil, pluginapi.Errorf(pluginapi.CodeUnauthorized, "bad token %v", call.Config.Tenant["token"])
}
site, _ := call.Config.Tenant["site"].(string)
return []pluginapi.Option{{Value: site + "/" + in.Query, Label: in.Field}}, nil
},
)
srv := httptest.NewServer(p.Handler())
t.Cleanup(srv.Close)
h := NewPluginFormsHandler(reg, ten, activate.NewInvoker(oneClient{client.New(srv.URL, nil, nil)}),
plugintest.NewMemRepo(), nil, nil, nil)
r := gin.New()
r.Use(middleware.ErrorHandler())
r.POST("/plugins/:id/options", func(c *gin.Context) {
c.Set(types.TenantIDContextKey.String(), uint64(7))
h.Options(c)
})
return r, settings
}
func TestPluginFormOptions(t *testing.T) {
r, settings := formsEngine(t)
post := func(body string) *httptest.ResponseRecorder {
return call(
r,
http.MethodPost,
"/plugins/acme.jira/options",
body,
map[string]string{"Content-Type": "application/json"},
)
}
// The form's unsaved site is used; its redacted token is the stored one.
w := post(
`{"name":"projects","field":"project","scope":"tenant","query":"q","values":{"site":"new","token":"***"}}`,
)
var resp struct {
Data pluginapi.OptionsOutput `json:"data"`
}
if w.Code != http.StatusOK || json.Unmarshal(w.Body.Bytes(), &resp) != nil ||
len(
resp.Data.Options,
) != 1 || resp.Data.Options[0].Value != "new/q" || resp.Data.Options[0].Label != "project" {
t.Fatalf("options = %d %s", w.Code, w.Body)
}
// A token the user typed replaces the stored one; the plugin's refusal
// reaches the form as a 400.
w = post(`{"name":"projects","field":"project","scope":"tenant","values":{"token":"typed"}}`)
if w.Code != http.StatusBadRequest ||
!strings.Contains(w.Body.String(), "unauthorized") {
t.Fatalf("typed token = %d %s", w.Code, w.Body)
}
if w := post(`{"name":"nope","field":"project","scope":"tenant"}`); w.Code != http.StatusNotFound {
t.Fatalf("unknown source = %d %s", w.Code, w.Body)
}
if w := post(`{"name":"projects","field":"project","scope":"system"}`); w.Code != http.StatusForbidden {
t.Fatalf("system scope for a workspace admin = %d", w.Code)
}
if w := post(`{"name":"projects","field":"project","scope":"other"}`); w.Code != http.StatusBadRequest {
t.Fatalf("bad scope = %d", w.Code)
}
_ = settings.Upsert(context.Background(),
&types.PluginTenantSetting{TenantID: 7, PluginID: "acme.jira", Enabled: false}, "enabled")
if w := post(`{"name":"projects","field":"project","scope":"tenant"}`); w.Code != http.StatusNotFound {
t.Fatalf("a switched-off plugin = %d", w.Code)
}
}
func TestPluginOAuthCallbackPage(t *testing.T) {
gin.SetMode(gin.TestMode)
reg := registry.New()
h := NewPluginFormsHandler(reg, nil, nil, nil, nil, nil,
pluginoauth.NewService(reg, plugintest.NewMemRepo(), nil, nil))
r := gin.New()
r.GET(pluginoauth.CallbackPath, h.OAuthCallback)
w := call(r, http.MethodGet, pluginoauth.CallbackPath+"?state=%3C%2Fscript%3E&code=x", "", nil)
body := w.Body.String()
if w.Code != http.StatusBadRequest || w.Header().Get("Cache-Control") != "no-store" ||
strings.Contains(body, "</script>'") || !strings.Contains(body, "ok: false") {
t.Fatalf("callback = %d %s", w.Code, body)
}
// Without a known state there is no origin to post the result to.
if !strings.Contains(body, `window.opener && ""`) {
t.Fatalf("callback posts to an origin: %s", body)
}
}
+95
View File
@@ -5,6 +5,8 @@ import (
"sync"
"time"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/plugin/configschema"
"github.com/Tencent/WeKnora/internal/plugin/install"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/plugin/tenancy"
@@ -41,6 +43,65 @@ type Invoker struct {
// webhooks and publicBase give plugins their workspace's webhook URLs.
webhooks *webhook.Tokens
publicBase string
oauth OAuthResolver
}
// OAuthResolver turns a form field's "oauth:<id>" into an access token.
type OAuthResolver interface {
AccessToken(ctx context.Context, pluginID string, tenantID uint64, ref string) (string, error)
}
// SetOAuth lets calls carry fresh access tokens in place of OAuth
// references.
func (iv *Invoker) SetOAuth(r OAuthResolver) {
iv.mu.Lock()
iv.oauth = r
iv.mu.Unlock()
}
// resolveOAuth returns a configuration with every "oauth:<id>" string
// replaced by the connection's access token, copying what it changes so a
// caller's cached configuration keeps its references. A connection that is
// gone or cannot be refreshed becomes "": the plugin sees no token and
// reports unauthorized.
func (iv *Invoker) resolveOAuth(
ctx context.Context, pluginID string, tenantID uint64, cfg map[string]any,
) map[string]any {
iv.mu.RLock()
r := iv.oauth
iv.mu.RUnlock()
if r == nil || cfg == nil {
return cfg
}
var walk func(v any) any
walk = func(v any) any {
switch t := v.(type) {
case string:
if !configschema.IsOAuthRef(t) {
return t
}
tok, err := r.AccessToken(ctx, pluginID, tenantID, t)
if err != nil {
logger.Warnf(ctx, "[plugin] %s: OAuth connection %s: %v", pluginID, t, err)
return ""
}
return tok
case map[string]any:
out := make(map[string]any, len(t))
for k, x := range t {
out[k] = walk(x)
}
return out
case []any:
out := make([]any, len(t))
for i, x := range t {
out[i] = walk(x)
}
return out
}
return v
}
return walk(cfg).(map[string]any)
}
// SetWebhooks lets calls carry the workspace's webhook URLs, when WeKnora's
@@ -154,6 +215,10 @@ func (iv *Invoker) Envelope(
}
env.Config.Tenant = cfg
}
// System connections belong to the platform (tenant 0).
env.Config.System = iv.resolveOAuth(ctx, m.ID, 0, env.Config.System)
env.Config.Tenant = iv.resolveOAuth(ctx, m.ID, env.Context.TenantID, env.Config.Tenant)
env.Config.Instance = iv.resolveOAuth(ctx, m.ID, env.Context.TenantID, env.Config.Instance)
return env, nil
}
@@ -196,6 +261,36 @@ func (iv *Invoker) Call(
return present(m.ID, c.Call(ctx, path, env, input, out))
}
// ConfigOverride replaces configuration scopes for one call: a form being
// edited asks the plugin with what it holds, not what is stored. A nil scope
// keeps the stored one.
type ConfigOverride struct {
System map[string]any
Tenant map[string]any
}
// CallOverriding is Call with some configuration scopes replaced.
func (iv *Invoker) CallOverriding(
ctx context.Context, m *manifest.Manifest, path string, instance map[string]any, override ConfigOverride,
input, out any,
) error {
c, err := iv.clients.Client(ctx, m)
if err != nil {
return present(m.ID, err)
}
env, err := iv.Envelope(ctx, m, instance)
if err != nil {
return err
}
if override.System != nil {
env.Config.System = iv.resolveOAuth(ctx, m.ID, 0, override.System)
}
if override.Tenant != nil {
env.Config.Tenant = iv.resolveOAuth(ctx, m.ID, env.Context.TenantID, override.Tenant)
}
return present(m.ID, c.Call(ctx, path, env, input, out))
}
// Stream makes a streaming call to a plugin.
func (iv *Invoker) Stream(
ctx context.Context, m *manifest.Manifest, path string, instance map[string]any, input any,
+91
View File
@@ -14,6 +14,7 @@ import (
"fmt"
"regexp"
"sort"
"strings"
)
// Types a schema node may declare.
@@ -75,6 +76,40 @@ type Schema struct {
Order int `json:"x-order,omitempty"`
// ModelTypes limits a model vendor field to some model types.
ModelTypes []string `json:"x-model-types,omitempty"`
// Options fills a field's choices from the plugin (its
// /v1/options/{name} endpoint), for choices that depend on the
// account: projects, spaces, channels.
Options *OptionsSource `json:"x-options,omitempty"`
// OAuth makes a string field an authorization: the form shows a
// connect button, WeKnora runs the OAuth code flow and the field holds
// a reference WeKnora swaps for a fresh access token at call time.
OAuth *OAuthSpec `json:"x-oauth,omitempty"`
}
// OptionsSource names where a field's choices come from.
type OptionsSource struct {
// Name is the plugin's options endpoint: /v1/options/{name}.
Name string `json:"name"`
// DependsOn lists dotted paths of fields whose values the choices
// depend on; the form reloads them when those change.
DependsOn []string `json:"dependsOn,omitempty"`
// Search asks the plugin with what the user types (large lists).
Search bool `json:"search,omitempty"`
}
// OAuthSpec is an OAuth 2.0 authorization code flow.
type OAuthSpec struct {
AuthorizeURL string `json:"authorizeUrl"`
TokenURL string `json:"tokenUrl"`
Scopes []string `json:"scopes,omitempty"`
// ClientID and ClientSecret are usually ${system.<key>}: the OAuth app
// is registered once for the platform.
ClientID string `json:"clientId"`
ClientSecret string `json:"clientSecret,omitempty"`
// PKCE adds a code challenge (S256).
PKCE bool `json:"pkce,omitempty"`
// Params are extra authorize parameters (audience, prompt).
Params map[string]string `json:"params,omitempty"`
}
// Object returns an empty object schema.
@@ -173,8 +208,44 @@ func (s *Schema) check(path string, errs *[]error) {
fail("oneOf entries must carry a const")
}
}
if o := s.Options; o != nil {
if !optionsNamePattern.MatchString(o.Name) {
fail("x-options.name must be lowercase letters, digits, '_' or '-'")
}
if s.Type != TypeString && (s.Type != TypeArray || s.Items == nil || s.Items.Type != TypeString) {
fail("x-options is only allowed on strings or arrays of strings")
}
}
if o := s.OAuth; o != nil {
if s.Type != TypeString {
fail("x-oauth is only allowed on strings")
}
if s.Secret {
fail("x-oauth fields hold a reference, not a secret; drop x-secret")
}
for name, u := range map[string]string{"authorizeUrl": o.AuthorizeURL, "tokenUrl": o.TokenURL} {
if !strings.HasPrefix(u, "https://") && !strings.HasPrefix(u, "http://localhost") {
fail("x-oauth.%s must be an https URL", name)
}
}
if o.ClientID == "" {
fail("x-oauth.clientId is required")
}
}
}
// OAuthRefPrefix starts the value of an x-oauth field: a reference to an
// OAuth connection WeKnora keeps (its tokens never reach the browser).
const OAuthRefPrefix = "oauth:"
var oauthRefPattern = regexp.MustCompile(`^oauth:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`)
// IsOAuthRef reports whether a value references an OAuth connection.
func IsOAuthRef(v string) bool { return oauthRefPattern.MatchString(v) }
// optionsNamePattern is an options endpoint name.
var optionsNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9_-]{0,63}$`)
func (s *Schema) containsSecret() bool {
if s.Secret {
return true
@@ -187,6 +258,26 @@ func (s *Schema) containsSecret() bool {
return s.Items != nil && s.Items.containsSecret()
}
// OAuthFields returns the x-oauth fields by dotted path. A nil schema has
// none.
func (s *Schema) OAuthFields() map[string]*OAuthSpec {
out := map[string]*OAuthSpec{}
if s == nil {
return out
}
var walk func(n *Schema, path string)
walk = func(n *Schema, path string) {
if n.OAuth != nil {
out[path] = n.OAuth
}
for key, p := range n.Properties {
walk(p, joinPath(path, key))
}
}
walk(s, "")
return out
}
// SecretPaths lists the dotted paths of every secret field, sorted.
func (s *Schema) SecretPaths() []string {
var out []string
+6
View File
@@ -139,6 +139,12 @@ func (vd *validator) value(s *Schema, v any, path string) {
add(CodeType, "must be a string")
return
}
if s.OAuth != nil {
if str != "" && !IsOAuthRef(str) {
add(CodeFormat, "must be connected through its OAuth button")
}
return
}
validateString(s, str, add)
case TypeNumber, TypeInteger:
n, ok := toFloat(v)
+40
View File
@@ -0,0 +1,40 @@
package host
import (
"os"
"strings"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
)
// envMemoryDefault is the memory limit of plugins that declare none
// (runtime.resources.memory), e.g. "1Gi". Unset: no limit. The cgroup
// setting (envCgroup) is Linux-only.
const envMemoryDefault = "WEKNORA_PLUGIN_MEMORY_DEFAULT"
// limits is what a plugin process may use; zero means no limit.
type limits struct {
cpuMilli int64
memoryBytes int64
}
// limitsFor reads a plugin's declared resources, with the platform default
// memory for plugins that declare none. The manifest was validated, so
// parse errors only mean "no limit".
func limitsFor(m *manifest.Manifest) limits {
var l limits
if r := m.Runtime.Resources; r != nil {
if r.CPU != "" {
l.cpuMilli, _ = manifest.ParseCPU(r.CPU)
}
if r.Memory != "" {
l.memoryBytes, _ = manifest.ParseMemory(r.Memory)
}
}
if l.memoryBytes == 0 {
if def := strings.TrimSpace(os.Getenv(envMemoryDefault)); def != "" {
l.memoryBytes, _ = manifest.ParseMemory(def)
}
}
return l
}
+73
View File
@@ -0,0 +1,73 @@
//go:build linux
package host
import (
"context"
"fmt"
"os"
"path/filepath"
"strconv"
"strings"
"golang.org/x/sys/unix"
"github.com/Tencent/WeKnora/internal/logger"
)
// cpuPeriod is the cgroup cpu.max period, in microseconds.
const cpuPeriod = 100000
// envCgroup names a cgroup v2 directory delegated to WeKnora. Each plugin
// process then runs in a child cgroup with memory.max and cpu.max set from
// its resources. Unset: memory is capped with an rlimit only and CPU is not
// capped.
const envCgroup = "WEKNORA_PLUGIN_CGROUP"
// applyLimits caps a started plugin process. Memory is capped with
// RLIMIT_DATA, which needs no privileges and covers the heap and anonymous
// mappings of Go, Python and Node; with a delegated cgroup, memory.max and
// cpu.max apply as well. Failures are logged: an unlimited plugin still
// runs. It returns a cleanup that removes the plugin's cgroup.
func applyLimits(pid int, id string, l limits) func() {
ctx := context.Background()
if l.memoryBytes > 0 {
lim := &unix.Rlimit{Cur: uint64(l.memoryBytes), Max: uint64(l.memoryBytes)}
if err := unix.Prlimit(pid, unix.RLIMIT_DATA, lim, nil); err != nil {
logger.Warnf(ctx, "[plugin] %s: cap memory: %v", id, err)
}
}
root := strings.TrimSpace(os.Getenv(envCgroup))
if root == "" || (l.memoryBytes == 0 && l.cpuMilli == 0) {
return func() {}
}
dir := filepath.Join(root, fmt.Sprintf("plugin-%d", pid))
if err := setupCgroup(dir, pid, l); err != nil {
logger.Warnf(ctx, "[plugin] %s: cgroup %s: %v", id, dir, err)
_ = os.Remove(dir)
return func() {}
}
return func() { _ = os.Remove(dir) }
}
func setupCgroup(dir string, pid int, l limits) error {
if err := os.Mkdir(dir, 0o755); err != nil && !os.IsExist(err) {
return err
}
write := func(file, value string) error {
return os.WriteFile(filepath.Join(dir, file), []byte(value), 0o644)
}
if l.memoryBytes > 0 {
if err := write("memory.max", strconv.FormatInt(l.memoryBytes, 10)); err != nil {
return err
}
}
if l.cpuMilli > 0 {
quota := l.cpuMilli * cpuPeriod / 1000
if err := write("cpu.max", fmt.Sprintf("%d %d", quota, cpuPeriod)); err != nil {
return err
}
}
return write("cgroup.procs", strconv.Itoa(pid))
}
+56
View File
@@ -0,0 +1,56 @@
//go:build linux
package host
import (
"fmt"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
)
func TestApplyLimitsCapsMemory(t *testing.T) {
cmd := exec.Command("sleep", "30")
if err := cmd.Start(); err != nil {
t.Skipf("no sleep: %v", err)
}
defer func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }()
// A cgroup root we cannot write to is logged, not fatal.
t.Setenv(envCgroup, filepath.Join(t.TempDir(), "not-a-cgroup"))
release := applyLimits(cmd.Process.Pid, "acme.test", limits{memoryBytes: 64 << 20, cpuMilli: 500})
defer release()
raw, err := os.ReadFile(fmt.Sprintf("/proc/%d/limits", cmd.Process.Pid))
if err != nil {
t.Fatal(err)
}
for _, line := range strings.Split(string(raw), "\n") {
if strings.HasPrefix(line, "Max data size") {
if !strings.Contains(line, fmt.Sprint(64<<20)) {
t.Fatalf("limit not applied: %s", line)
}
return
}
}
t.Fatal("no data size limit line")
}
func TestSetupCgroupWritesLimits(t *testing.T) {
// A plain directory stands in for a delegated cgroup: the files are
// what the kernel would read.
dir := filepath.Join(t.TempDir(), "plugin-1")
if err := setupCgroup(dir, 4242, limits{memoryBytes: 1 << 30, cpuMilli: 1500}); err != nil {
t.Fatal(err)
}
for file, want := range map[string]string{
"memory.max": fmt.Sprint(1 << 30), "cpu.max": "150000 100000", "cgroup.procs": "4242",
} {
got, _ := os.ReadFile(filepath.Join(dir, file))
if string(got) != want {
t.Errorf("%s = %q, want %q", file, got, want)
}
}
}
+6
View File
@@ -0,0 +1,6 @@
//go:build !linux
package host
// applyLimits is Linux only: elsewhere plugins run uncapped (desktop, dev).
func applyLimits(int, string, limits) func() { return func() {} }
+25
View File
@@ -0,0 +1,25 @@
package host
import (
"testing"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
)
func TestLimitsFor(t *testing.T) {
m := &manifest.Manifest{Runtime: manifest.Runtime{Resources: &manifest.Resources{CPU: "250m", Memory: "256Mi"}}}
if l := limitsFor(m); l.cpuMilli != 250 || l.memoryBytes != 256<<20 {
t.Fatalf("declared = %+v", l)
}
bare := &manifest.Manifest{}
if l := limitsFor(bare); l != (limits{}) {
t.Fatalf("no resources, no default = %+v", l)
}
t.Setenv(envMemoryDefault, "1Gi")
if l := limitsFor(bare); l.memoryBytes != 1<<30 || l.cpuMilli != 0 {
t.Fatalf("platform default = %+v", l)
}
if l := limitsFor(m); l.memoryBytes != 256<<20 {
t.Fatalf("a declared limit wins over the default: %+v", l)
}
}
+6 -1
View File
@@ -278,12 +278,17 @@ func (p *process) launch(ctx context.Context, entry string) (*launched, error) {
return nil, fmt.Errorf("start %s: %w", filepath.Base(entry), err)
}
id := p.spec.m.ID + "@" + p.spec.m.Version
releaseLimits := applyLimits(cmd.Process.Pid, id, limitsFor(p.spec.m))
handshake := make(chan pluginapi.Handshake, 1)
hsErr := make(chan error, 1)
go forwardLogs(stderr, id, nil, nil)
go forwardLogs(stdout, id, handshake, hsErr)
exited := make(chan error, 1)
go func() { exited <- cmd.Wait() }()
go func() {
err := cmd.Wait()
releaseLimits()
exited <- err
}()
kill := func() {
stopChild(cmd, exited)
+15
View File
@@ -155,6 +155,9 @@ type Contribution struct {
MCP *MCPServer `json:"mcp,omitempty" yaml:"mcp"`
// FileTypes are the lower-case extensions a parser handles ("pdf").
FileTypes []string `json:"fileTypes,omitempty" yaml:"fileTypes"`
// InstanceSchemaJSON is the instanceSchema file's content as JSON,
// filled when the package is opened so every node can read it.
InstanceSchemaJSON json.RawMessage `json:"instanceSchemaJson,omitempty" yaml:"-"`
// Entry is the HTML page of a UI contribution (pages, settingsSections,
// kbTabs), a path under UIRoot.
Entry string `json:"entry,omitempty" yaml:"entry"`
@@ -285,6 +288,18 @@ func (m *Manifest) validateRuntime(add func(string, ...any)) {
if m.Runtime.Entry == "" {
add("runtime.entry is required for host plugins")
}
if r := m.Runtime.Resources; r != nil {
if r.CPU != "" {
if _, err := ParseCPU(r.CPU); err != nil {
add("runtime.resources.%v", err)
}
}
if r.Memory != "" {
if _, err := ParseMemory(r.Memory); err != nil {
add("runtime.resources.%v", err)
}
}
}
default:
add("runtime.type %q is not one of builtin, declarative, host, remote, kubernetes", m.Runtime.Type)
}
+21
View File
@@ -244,3 +244,24 @@ func TestUIContributions(t *testing.T) {
t.Fatal("min role defaults")
}
}
func TestResourceAmounts(t *testing.T) {
for in, want := range map[string]int64{"500m": 500, "0.5": 500, "2": 2000} {
if got, err := ParseCPU(in); err != nil || got != want {
t.Errorf("ParseCPU(%q) = %d, %v", in, got, err)
}
}
for in, want := range map[string]int64{"512Mi": 512 << 20, "1Gi": 1 << 30, "500M": 500e6, "1024": 1024} {
if got, err := ParseMemory(in); err != nil || got != want {
t.Errorf("ParseMemory(%q) = %d, %v", in, got, err)
}
}
for _, bad := range []string{"", "-1", "lots", "0m"} {
if _, err := ParseCPU(bad); err == nil {
t.Errorf("ParseCPU(%q) accepted", bad)
}
if _, err := ParseMemory(bad); err == nil {
t.Errorf("ParseMemory(%q) accepted", bad)
}
}
}
+59
View File
@@ -0,0 +1,59 @@
package manifest
import (
"fmt"
"strconv"
"strings"
)
// ParseCPU reads a CPU amount as Kubernetes writes it ("500m", "0.5", "2")
// and returns it in millicores.
func ParseCPU(s string) (int64, error) {
s = strings.TrimSpace(s)
if strings.HasSuffix(s, "m") {
n, err := strconv.ParseInt(strings.TrimSuffix(s, "m"), 10, 64)
if err != nil || n <= 0 {
return 0, fmt.Errorf("cpu %q must be millicores like 500m", s)
}
return n, nil
}
f, err := strconv.ParseFloat(s, 64)
if err != nil || f <= 0 {
return 0, fmt.Errorf("cpu %q must be cores like 0.5 or millicores like 500m", s)
}
return int64(f * 1000), nil
}
var memUnits = []struct {
suffix string
factor int64
}{
{"Ki", 1 << 10},
{"Mi", 1 << 20},
{"Gi", 1 << 30},
{"Ti", 1 << 40},
{"K", 1e3},
{"M", 1e6},
{"G", 1e9},
{"T", 1e12},
}
// ParseMemory reads a memory amount as Kubernetes writes it ("512Mi",
// "1Gi", "500M") and returns bytes.
func ParseMemory(s string) (int64, error) {
s = strings.TrimSpace(s)
for _, u := range memUnits {
if strings.HasSuffix(s, u.suffix) {
n, err := strconv.ParseInt(strings.TrimSuffix(s, u.suffix), 10, 64)
if err != nil || n <= 0 {
break
}
return n * u.factor, nil
}
}
n, err := strconv.ParseInt(s, 10, 64)
if err != nil || n <= 0 {
return 0, fmt.Errorf("memory %q must be an amount like 512Mi or 1Gi", s)
}
return n, nil
}
+332
View File
@@ -0,0 +1,332 @@
// Package oauth runs the OAuth 2.0 authorization code flow behind plugin form
// fields that declare x-oauth. The platform's OAuth app (client ID and
// secret in the plugin's system configuration) authorizes an account; the
// tokens are kept sealed in plugin_oauth_connections and the field holds
// only "oauth:<connection id>". When WeKnora calls the plugin it swaps the
// reference for a fresh access token, refreshing it when it is about to
// expire.
package oauth
import (
"context"
"crypto/rand"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"net/http"
"strings"
"time"
"github.com/google/uuid"
"github.com/redis/go-redis/v9"
"golang.org/x/oauth2"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/plugin/configschema"
"github.com/Tencent/WeKnora/internal/plugin/install"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/plugin/registry"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
"github.com/Tencent/WeKnora/internal/utils"
)
// CallbackPath is where authorization servers send the browser back.
const CallbackPath = "/api/v1/plugin-oauth/callback"
// refreshBefore refreshes tokens this close to expiring.
const refreshBefore = time.Minute
// Scopes a form edits, as pluginapi.OptionsScope*.
const (
ScopeSystem = "system"
ScopeTenant = "tenant"
ScopeInstance = "instance"
)
// ErrNotConnected means a reference names no usable connection.
var ErrNotConnected = errors.New("the OAuth connection is gone; connect again")
// Target is the form field an authorization is for.
type Target struct {
PluginID string
// TenantID is 0 for the system configuration.
TenantID uint64
Scope string
Contribution string // "<point>/<id>" for instance forms
Field string // dotted path
}
// Service runs the flow and resolves references.
type Service struct {
registry *registry.Registry
plugins interfaces.PluginRepository
repo interfaces.PluginOAuthRepository
states *stateStore
http *http.Client
now func() time.Time
}
// NewService creates the service; rdb may be nil (single node).
func NewService(
reg *registry.Registry, plugins interfaces.PluginRepository, repo interfaces.PluginOAuthRepository,
rdb *redis.Client,
) *Service {
cfg := utils.DefaultSSRFSafeHTTPClientConfig()
cfg.Timeout = 30 * time.Second
return &Service{
registry: reg, plugins: plugins, repo: repo, states: newStateStore(rdb),
http: utils.NewSSRFSafeHTTPClient(cfg), now: time.Now,
}
}
// spec finds the x-oauth spec of a target's field.
func (s *Service) spec(m *manifest.Manifest, t Target) (*configschema.OAuthSpec, error) {
var raw json.RawMessage
switch t.Scope {
case ScopeSystem:
raw = m.Config.SystemSchema
case ScopeTenant:
raw = m.Config.TenantSchema
case ScopeInstance:
point, id, _ := strings.Cut(t.Contribution, "/")
for _, c := range m.Contributes[manifest.Point(point)] {
if c.ID == id {
raw = c.InstanceSchemaJSON
}
}
}
if len(raw) == 0 {
return nil, fmt.Errorf("no %s form %q in %s", t.Scope, t.Contribution, m.ID)
}
schema, err := configschema.Parse(raw)
if err != nil {
return nil, err
}
spec, ok := schema.OAuthFields()[t.Field]
if !ok {
return nil, fmt.Errorf("field %s has no x-oauth", t.Field)
}
return spec, nil
}
// config builds the OAuth client, with its credentials from the plugin's
// system configuration.
func (s *Service) config(ctx context.Context, m *manifest.Manifest, spec *configschema.OAuthSpec, redirect string) (
*oauth2.Config, error,
) {
system, _, err := install.OpenSystemConfig(ctx, s.plugins, m)
if err != nil {
return nil, err
}
lookup := func(scope, key string) (string, bool) {
if scope != manifest.ScopeSystem {
return "", false
}
v, ok := system[key].(string)
return v, ok && v != ""
}
id, err := manifest.ExpandTemplate(spec.ClientID, lookup)
if err != nil {
return nil, fmt.Errorf("the platform has not set up this plugin's OAuth app: %w", err)
}
secret := ""
if spec.ClientSecret != "" {
if secret, err = manifest.ExpandTemplate(spec.ClientSecret, lookup); err != nil {
return nil, fmt.Errorf("the platform has not set up this plugin's OAuth app: %w", err)
}
}
return &oauth2.Config{
ClientID: id, ClientSecret: secret, Scopes: spec.Scopes, RedirectURL: redirect,
Endpoint: oauth2.Endpoint{AuthURL: spec.AuthorizeURL, TokenURL: spec.TokenURL},
}, nil
}
func randomToken() string {
b := make([]byte, 32)
_, _ = rand.Read(b)
return base64.RawURLEncoding.EncodeToString(b)
}
// Start begins an authorization for a field and returns the URL to open and
// the state that will come back. origin is the app's origin, where the
// callback page reports the result; baseURL is where the authorization
// server reaches WeKnora.
func (s *Service) Start(ctx context.Context, t Target, userID, origin, baseURL string) (string, string, error) {
m, ok := s.registry.Plugin(t.PluginID)
if !ok {
return "", "", fmt.Errorf("plugin %s is not loaded", t.PluginID)
}
spec, err := s.spec(m, t)
if err != nil {
return "", "", err
}
redirect := strings.TrimSuffix(baseURL, "/") + CallbackPath
cfg, err := s.config(ctx, m, spec, redirect)
if err != nil {
return "", "", err
}
state := randomToken()
st := pending{Target: t, UserID: userID, Origin: origin, Redirect: redirect}
var opts []oauth2.AuthCodeOption
if spec.PKCE {
st.Verifier = oauth2.GenerateVerifier()
opts = append(opts, oauth2.S256ChallengeOption(st.Verifier))
}
for k, v := range spec.Params {
opts = append(opts, oauth2.SetAuthURLParam(k, v))
}
opts = append(opts, oauth2.AccessTypeOffline)
if err := s.states.put(ctx, state, st); err != nil {
return "", "", err
}
return cfg.AuthCodeURL(state, opts...), state, nil
}
// Result is how an authorization ended, for the callback page.
type Result struct {
State string
Origin string
ConnectionID string
Err error
}
// Complete exchanges the code the authorization server sent back and keeps
// the tokens as a new connection.
func (s *Service) Complete(ctx context.Context, state, code, errParam string) Result {
st, ok, err := s.states.take(ctx, state)
if err != nil || !ok {
return Result{State: state, Err: errors.New("this authorization expired or was already used; try again")}
}
res := Result{State: state, Origin: st.Origin}
if errParam != "" {
res.Err = fmt.Errorf("the authorization was refused: %s", errParam)
return res
}
m, ok := s.registry.Plugin(st.Target.PluginID)
if !ok {
res.Err = fmt.Errorf("plugin %s is not loaded", st.Target.PluginID)
return res
}
spec, err := s.spec(m, st.Target)
if err == nil {
var cfg *oauth2.Config
if cfg, err = s.config(ctx, m, spec, st.Redirect); err == nil {
var opts []oauth2.AuthCodeOption
if st.Verifier != "" {
opts = append(opts, oauth2.VerifierOption(st.Verifier))
}
var tok *oauth2.Token
if tok, err = cfg.Exchange(context.WithValue(ctx, oauth2.HTTPClient, s.http), code, opts...); err == nil {
res.ConnectionID, err = s.save(ctx, st, tok)
}
}
}
if err != nil {
logger.Warnf(ctx, "[plugin] OAuth for %s %s: %v", st.Target.PluginID, st.Target.Field, err)
res.Err = err
}
return res
}
func (s *Service) save(ctx context.Context, st pending, tok *oauth2.Token) (string, error) {
sealed, err := seal(tok)
if err != nil {
return "", err
}
now := s.now()
c := &types.PluginOAuthConnection{
ID: uuid.NewString(), PluginID: st.Target.PluginID, TenantID: st.Target.TenantID, Scope: st.Target.Scope,
Contribution: st.Target.Contribution, Field: st.Target.Field, Token: sealed, CreatedBy: st.UserID,
CreatedAt: now, UpdatedAt: now,
}
if !tok.Expiry.IsZero() {
exp := tok.Expiry
c.ExpiresAt = &exp
}
return c.ID, s.repo.Create(ctx, c)
}
func seal(tok *oauth2.Token) (string, error) {
raw, err := json.Marshal(tok)
if err != nil {
return "", err
}
return utils.EncryptAESGCM(string(raw), utils.GetAESKey())
}
func open(sealed string) (*oauth2.Token, error) {
raw, err := utils.DecryptStoredSecret(sealed)
if err != nil {
return nil, err
}
var tok oauth2.Token
return &tok, json.Unmarshal([]byte(raw), &tok)
}
// AccessToken resolves a field value "oauth:<id>" of a plugin in a tenant to
// a current access token, refreshing it when it is about to expire.
func (s *Service) AccessToken(ctx context.Context, pluginID string, tenantID uint64, ref string) (string, error) {
id := strings.TrimPrefix(ref, configschema.OAuthRefPrefix)
c, err := s.repo.Get(ctx, pluginID, tenantID, id)
if err != nil {
return "", err
}
if c == nil {
return "", ErrNotConnected
}
tok, err := open(c.Token)
if err != nil {
return "", err
}
if tok.Expiry.IsZero() || tok.Expiry.After(s.now().Add(refreshBefore)) {
return tok.AccessToken, nil
}
if tok.RefreshToken == "" {
return "", ErrNotConnected
}
m, ok := s.registry.Plugin(pluginID)
if !ok {
return "", ErrNotConnected
}
spec, err := s.spec(m, Target{PluginID: pluginID, Scope: c.Scope, Contribution: c.Contribution, Field: c.Field})
if err != nil {
return "", err
}
cfg, err := s.config(ctx, m, spec, "")
if err != nil {
return "", err
}
// A token source only refreshes an invalid token; ours may still be
// valid for seconds.
stale := *tok
stale.AccessToken = ""
fresh, err := cfg.TokenSource(context.WithValue(ctx, oauth2.HTTPClient, s.http), &stale).Token()
if err != nil {
return "", fmt.Errorf("refresh the OAuth token: %w", err)
}
if fresh.RefreshToken == "" {
fresh.RefreshToken = tok.RefreshToken
}
sealed, err := seal(fresh)
if err != nil {
return "", err
}
prev := c.UpdatedAt
c.Token, c.UpdatedAt = sealed, s.now()
if !fresh.Expiry.IsZero() {
exp := fresh.Expiry
c.ExpiresAt = &exp
}
saved, err := s.repo.UpdateToken(ctx, c, prev)
if err != nil {
return "", err
}
if !saved {
// Another node refreshed first; its token is the current one (a
// rotated refresh token makes ours unusable).
return s.AccessToken(ctx, pluginID, tenantID, ref)
}
return fresh.AccessToken, nil
}
+198
View File
@@ -0,0 +1,198 @@
package oauth
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/Tencent/WeKnora/internal/plugin/configschema"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/plugin/plugintest"
"github.com/Tencent/WeKnora/internal/plugin/registry"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/utils"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
type memConnections struct {
mu sync.Mutex
rows map[string]types.PluginOAuthConnection
}
func (m *memConnections) Create(_ context.Context, c *types.PluginOAuthConnection) error {
m.mu.Lock()
defer m.mu.Unlock()
m.rows[c.ID] = *c
return nil
}
func (m *memConnections) Get(
_ context.Context,
pluginID string,
tenantID uint64,
id string,
) (*types.PluginOAuthConnection, error) {
m.mu.Lock()
defer m.mu.Unlock()
c, ok := m.rows[id]
if !ok || c.PluginID != pluginID || c.TenantID != tenantID {
return nil, nil
}
return &c, nil
}
func (m *memConnections) UpdateToken(_ context.Context, c *types.PluginOAuthConnection, prev time.Time) (bool, error) {
m.mu.Lock()
defer m.mu.Unlock()
if !m.rows[c.ID].UpdatedAt.Equal(prev) {
return false, nil
}
m.rows[c.ID] = *c
return true, nil
}
// provider is a fake authorization server.
type provider struct {
*httptest.Server
refreshes atomic.Int32
verifier string
}
func newProvider(t *testing.T) *provider {
p := &provider{}
p.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_ = r.ParseForm()
if user, pass, _ := r.BasicAuth(); user != "app-id" || pass != "app-secret" {
if r.Form.Get("client_id") != "app-id" || r.Form.Get("client_secret") != "app-secret" {
http.Error(w, "bad client", http.StatusUnauthorized)
return
}
}
w.Header().Set("Content-Type", "application/json")
switch r.Form.Get("grant_type") {
case "authorization_code":
if r.Form.Get("code") != "the-code" {
http.Error(w, `{"error":"invalid_grant"}`, http.StatusBadRequest)
return
}
p.verifier = r.Form.Get("code_verifier")
_, _ = fmt.Fprint(w, `{"access_token":"at-1","refresh_token":"rt-1","token_type":"bearer","expires_in":30}`)
case "refresh_token":
n := p.refreshes.Add(1)
_, _ = fmt.Fprintf(w, `{"access_token":"at-%d","token_type":"bearer","expires_in":3600}`, n+1)
}
}))
t.Cleanup(p.Close)
return p
}
func setup(t *testing.T, p *provider) *Service {
t.Helper()
utils.SetSSRFWhitelistFromRaw("127.0.0.1,localhost")
t.Cleanup(func() { utils.SetSSRFWhitelistFromRaw("") })
tenantSchema, _ := json.Marshal(map[string]any{
"type": "object",
"properties": map[string]any{"account": map[string]any{
"type": "string", "x-oauth": map[string]any{
"authorizeUrl": "https://auth.example.com/authorize",
"tokenUrl": strings.Replace(p.URL, "127.0.0.1", "localhost", 1) + "/token",
"scopes": []string{
"read",
}, "clientId": "${system.client_id}", "clientSecret": "${system.client_secret}",
"pkce": true, "params": map[string]string{"audience": "api.example.com"},
},
}},
})
systemSchema, _ := json.Marshal(map[string]any{"type": "object", "properties": map[string]any{
"client_id": map[string]any{
"type": "string",
}, "client_secret": map[string]any{"type": "string", "x-secret": true},
}})
m := &manifest.Manifest{
SchemaVersion: manifest.SchemaVersion, ID: "acme.jira", Version: "1.0.0", APIVersion: pluginapi.APIVersion,
Name: manifest.Text("Jira", nil), Publisher: manifest.Publisher{ID: "acme"},
Runtime: manifest.Runtime{Type: manifest.RuntimeHost, Kind: "binary", Entry: "bin/x"},
Contributes: manifest.Contributions{manifest.PointWebSearch: {{ID: "x", Name: manifest.Text("X", nil)}}},
Config: manifest.ConfigSchemas{SystemSchema: systemSchema, TenantSchema: tenantSchema},
}
reg := registry.New()
if err := reg.Register(m); err != nil {
t.Fatal(err)
}
plugins := plugintest.NewMemRepo()
sys, _ := json.Marshal(map[string]any{"client_id": "app-id", "client_secret": "app-secret"})
_ = plugins.SavePlugin(context.Background(), &types.InstalledPlugin{ID: "acme.jira", SystemConfig: types.JSON(sys)})
return NewService(reg, plugins, &memConnections{rows: map[string]types.PluginOAuthConnection{}}, nil)
}
func TestAuthorizeExchangeAndRefresh(t *testing.T) {
ctx := context.Background()
p := newProvider(t)
s := setup(t, p)
target := Target{PluginID: "acme.jira", TenantID: 7, Scope: ScopeTenant, Field: "account"}
authURL, state, err := s.Start(ctx, target, "u1", "https://app.example.com", "https://weknora.example.com")
if err != nil {
t.Fatal(err)
}
u, _ := url.Parse(authURL)
q := u.Query()
if u.Host != "auth.example.com" || q.Get("client_id") != "app-id" || q.Get("state") != state ||
q.Get("code_challenge_method") != "S256" || q.Get("audience") != "api.example.com" ||
q.Get("redirect_uri") != "https://weknora.example.com"+CallbackPath || q.Get("scope") != "read" {
t.Fatalf("authorize URL = %s", authURL)
}
res := s.Complete(ctx, state, "the-code", "")
if res.Err != nil || res.Origin != "https://app.example.com" || res.ConnectionID == "" || p.verifier == "" {
t.Fatalf("complete = %+v verifier %q", res, p.verifier)
}
if again := s.Complete(ctx, state, "the-code", ""); again.Err == nil {
t.Fatal("a state must not be usable twice")
}
ref := configschema.OAuthRefPrefix + res.ConnectionID
if !configschema.IsOAuthRef(ref) {
t.Fatalf("ref %s is not a reference", ref)
}
// The first token expires within the refresh window: it is refreshed,
// keeping the refresh token.
tok, err := s.AccessToken(ctx, "acme.jira", 7, ref)
if err != nil || tok != "at-2" || p.refreshes.Load() != 1 {
t.Fatalf("token = %q, %v (refreshes %d)", tok, err, p.refreshes.Load())
}
if tok, _ := s.AccessToken(ctx, "acme.jira", 7, ref); tok != "at-2" || p.refreshes.Load() != 1 {
t.Fatalf("a fresh token is reused, got %q", tok)
}
if _, err := s.AccessToken(ctx, "acme.jira", 8, ref); err == nil {
t.Fatal("another workspace cannot use the connection")
}
if raw := s.repo.(*memConnections).rows[res.ConnectionID].Token; strings.Contains(raw, "at-") &&
utils.GetAESKey() != nil {
t.Fatal("tokens are stored sealed")
}
}
func TestStartRefusesFieldsWithoutOAuth(t *testing.T) {
s := setup(t, newProvider(t))
for _, target := range []Target{
{PluginID: "acme.jira", TenantID: 7, Scope: ScopeTenant, Field: "other"},
{PluginID: "acme.jira", Scope: ScopeSystem, Field: "account"},
{PluginID: "acme.nope", TenantID: 7, Scope: ScopeTenant, Field: "account"},
} {
if _, _, err := s.Start(context.Background(), target, "u", "https://a", "https://a"); err == nil {
t.Errorf("%+v started", target)
}
}
if res := s.Complete(context.Background(), "unknown", "code", ""); res.Err == nil {
t.Fatal("an unknown state completed")
}
}
+94
View File
@@ -0,0 +1,94 @@
package oauth
import (
"context"
"encoding/json"
"errors"
"os"
"strings"
"sync"
"time"
"github.com/redis/go-redis/v9"
)
// stateTTL bounds an authorization from start to callback.
const stateTTL = 10 * time.Minute
// pending is an authorization in flight. It holds the PKCE verifier, a
// secret, so it lives server side and the state parameter is a random key.
type pending struct {
Target Target `json:"target"`
UserID string `json:"userId"`
Origin string `json:"origin"`
Redirect string `json:"redirect"`
Verifier string `json:"verifier,omitempty"`
}
// stateStore keeps pending authorizations in Redis, so the callback may land
// on any node, or in memory on a single node.
type stateStore struct {
rdb *redis.Client
mu sync.Mutex
mem map[string]memState
}
type memState struct {
value pending
expires time.Time
}
func newStateStore(rdb *redis.Client) *stateStore {
return &stateStore{rdb: rdb, mem: map[string]memState{}}
}
func stateKey(state string) string {
if ns := strings.TrimSpace(os.Getenv("WEKNORA_REDIS_NAMESPACE")); ns != "" {
return "weknora:plugin_oauth_state:" + ns + ":" + state
}
return "weknora:plugin_oauth_state:" + state
}
func (s *stateStore) put(ctx context.Context, state string, p pending) error {
if s.rdb != nil {
raw, err := json.Marshal(p)
if err != nil {
return err
}
return s.rdb.Set(ctx, stateKey(state), raw, stateTTL).Err()
}
s.mu.Lock()
defer s.mu.Unlock()
now := time.Now()
for k, v := range s.mem {
if now.After(v.expires) {
delete(s.mem, k)
}
}
s.mem[state] = memState{value: p, expires: now.Add(stateTTL)}
return nil
}
// take returns a pending authorization once: a state cannot be replayed.
func (s *stateStore) take(ctx context.Context, state string) (pending, bool, error) {
if s.rdb != nil {
raw, err := s.rdb.GetDel(ctx, stateKey(state)).Bytes()
if errors.Is(err, redis.Nil) {
return pending{}, false, nil
}
if err != nil {
return pending{}, false, err
}
var p pending
return p, true, json.Unmarshal(raw, &p)
}
s.mu.Lock()
defer s.mu.Unlock()
v, ok := s.mem[state]
delete(s.mem, state)
if !ok || time.Now().After(v.expires) {
return pending{}, false, nil
}
return v.value, true, nil
}
+30
View File
@@ -134,6 +134,20 @@ func (p *Package) loadConfigSchemas() error {
return errors.Join(errs...)
}
m.Config.SystemSchema, m.Config.TenantSchema = systemRaw, tenantRaw
schemas := []*configschema.Schema{system, tenant}
for point, list := range m.Contributes {
for i := range list {
if list[i].InstanceSchema == "" {
continue
}
s, raw := load(list[i].InstanceSchema)
m.Contributes[point][i].InstanceSchemaJSON = raw
schemas = append(schemas, s)
}
}
if len(errs) > 0 {
return errors.Join(errs...)
}
declared := func(s *configschema.Schema, key string) bool {
return s != nil && s.Properties[key] != nil
@@ -158,6 +172,22 @@ func (p *Package) loadConfigSchemas() error {
}
}
}
// An OAuth app is the platform's: its client ID and secret come from the
// system configuration.
for _, s := range schemas {
for path, spec := range s.OAuthFields() {
for _, v := range []string{spec.ClientID, spec.ClientSecret} {
for _, ref := range manifest.TemplateRefs(v) {
if ref.Scope != manifest.ScopeSystem || !declared(system, ref.Key) {
errs = append(errs, fmt.Errorf(
"%s: x-oauth client credentials may only use ${system.<key>} the config.system "+
"schema declares, not ${%s.%s}",
path, ref.Scope, ref.Key))
}
}
}
}
}
return errors.Join(errs...)
}
+9 -1
View File
@@ -21,6 +21,7 @@ import (
"github.com/Tencent/WeKnora/internal/mcpserver"
"github.com/Tencent/WeKnora/internal/middleware"
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
pluginoauth "github.com/Tencent/WeKnora/internal/plugin/oauth"
"github.com/Tencent/WeKnora/internal/plugin/webhook"
"github.com/Tencent/WeKnora/internal/tracing/langfuse"
"github.com/Tencent/WeKnora/internal/types/interfaces"
@@ -97,6 +98,7 @@ type RouterParams struct {
PluginHostAPI *hostapi.Handler
PluginUIHandler *handler.PluginUIHandler
PluginWebhookHandler *handler.PluginWebhookHandler
PluginFormsHandler *handler.PluginFormsHandler
WeKnoraCloudHandler *handler.WeKnoraCloudHandler
WikiPageHandler *handler.WikiPageHandler
MemoryHandler *handler.MemoryHandler
@@ -223,6 +225,11 @@ func NewRouter(params RouterParams) *gin.Engine {
}
// Plugin webhooks: third parties call a secret per-workspace URL; the
// handler checks it, so no login applies.
// OAuth for plugin form fields: authorization servers return here
// without a WeKnora login; the state names the pending authorization.
if params.PluginFormsHandler != nil {
r.GET(pluginoauth.CallbackPath, params.PluginFormsHandler.OAuthCallback)
}
if params.PluginWebhookHandler != nil {
hooks := r.Group(webhook.PathPrefix)
hooks.Any("/:id/:hook/:token", params.PluginWebhookHandler.Receive)
@@ -348,7 +355,8 @@ func NewRouter(params RouterParams) *gin.Engine {
RegisterEmbedChannelRoutes(v1, params.EmbedChannelHandler, rbacGuards)
RegisterMCPEndpointRoutes(v1, params.MCPEndpointHandler, rbacGuards)
RegisterDataSourceRoutes(v1, params.DataSourceHandler, params.DataSourceCredentialsHandler, rbacGuards)
RegisterPluginRoutes(v1, params.PluginHandler, params.PluginUIHandler, params.PluginWebhookHandler, rbacGuards)
RegisterPluginRoutes(v1, params.PluginHandler, params.PluginUIHandler, params.PluginWebhookHandler,
params.PluginFormsHandler, rbacGuards)
RegisterPluginAdminRoutes(v1, params.PluginAdminHandler, rbacGuards)
RegisterWeKnoraCloudRoutes(v1, params.WeKnoraCloudHandler, rbacGuards)
RegisterWikiPageRoutes(v1, params.WikiPageHandler, rbacGuards)
+7 -1
View File
@@ -14,7 +14,7 @@ import (
// API keys.
func RegisterPluginRoutes(
r *gin.RouterGroup, h *handler.PluginHandler, ui *handler.PluginUIHandler, hooks *handler.PluginWebhookHandler,
g *rbacGuards,
forms *handler.PluginFormsHandler, g *rbacGuards,
) {
plugins := g.apiKeyGroup(r.Group("/plugins"), apiKeyFullAccess())
{
@@ -37,6 +37,12 @@ func RegisterPluginRoutes(
if hooks != nil {
plugins.GET("/:id/webhooks", g.Admin(), hooks.List)
}
// Plugin forms are filled by workspace admins, or system admins
// for the platform configuration (checked per request).
if forms != nil {
plugins.POST("/:id/options", g.AdminOrSystemAdmin(), forms.Options)
plugins.POST("/:id/oauth/start", g.AdminOrSystemAdmin(), forms.OAuthStart)
}
}
}
+11
View File
@@ -65,3 +65,14 @@ type PluginKVRepository interface {
// DeleteExpired removes entries that expired before now.
DeleteExpired(ctx context.Context, now time.Time) (int64, error)
}
// PluginOAuthRepository stores plugin OAuth connections.
type PluginOAuthRepository interface {
Create(ctx context.Context, c *types.PluginOAuthConnection) error
// Get returns (nil, nil) when the connection does not exist for this
// plugin and tenant.
Get(ctx context.Context, pluginID string, tenantID uint64, id string) (*types.PluginOAuthConnection, error)
// UpdateToken saves a refreshed token unless another node refreshed it
// since prev (the UpdatedAt read); it reports whether it saved.
UpdateToken(ctx context.Context, c *types.PluginOAuthConnection, prev time.Time) (bool, error)
}
+23
View File
@@ -66,3 +66,26 @@ type PluginKV struct {
// TableName pins the table so GORM's pluralizer cannot drift.
func (PluginKV) TableName() string { return "plugin_kv" }
// PluginOAuthConnection is an authorization a plugin form field made with
// its OAuth button (x-oauth). The field holds "oauth:<ID>"; the tokens stay
// here, sealed (enc:v1), and never reach the browser.
type PluginOAuthConnection struct {
ID string `json:"id" gorm:"type:varchar(36);primaryKey"`
PluginID string `json:"pluginId" gorm:"type:varchar(128);index:idx_plugin_oauth_connections_owner"`
// TenantID is 0 for the platform (system configuration).
TenantID uint64 `json:"-" gorm:"index:idx_plugin_oauth_connections_owner"`
// Scope, Contribution and Field find the x-oauth spec again when the
// token needs refreshing.
Scope string `json:"scope" gorm:"type:varchar(16)"`
Contribution string `json:"contribution" gorm:"type:varchar(160)"`
Field string `json:"field" gorm:"type:varchar(256)"`
Token string `json:"-" gorm:"type:text"`
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
CreatedBy string `json:"-" gorm:"type:varchar(64)"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
// TableName pins the table so GORM's pluralizer cannot drift.
func (PluginOAuthConnection) TableName() string { return "plugin_oauth_connections" }
@@ -0,0 +1 @@
DROP TABLE IF EXISTS plugin_oauth_connections;
@@ -0,0 +1,16 @@
-- OAuth connections of plugin forms (versioned 000119).
CREATE TABLE IF NOT EXISTS plugin_oauth_connections (
id VARCHAR(36) PRIMARY KEY,
plugin_id VARCHAR(128) NOT NULL,
tenant_id BIGINT NOT NULL,
scope VARCHAR(16) NOT NULL,
contribution VARCHAR(160) NOT NULL DEFAULT '',
field VARCHAR(256) NOT NULL,
token TEXT NOT NULL,
expires_at DATETIME,
created_by VARCHAR(64) NOT NULL DEFAULT '',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_plugin_oauth_connections_owner ON plugin_oauth_connections (plugin_id, tenant_id);
@@ -0,0 +1 @@
DROP TABLE IF EXISTS plugin_oauth_connections;
@@ -0,0 +1,22 @@
-- Migration 000119: OAuth connections of plugin forms (x-oauth). A form field
-- holds "oauth:<id>"; the tokens stay here, sealed, and WeKnora refreshes
-- them and hands the plugin a fresh access token at call time.
DO $$ BEGIN RAISE NOTICE '[Migration 000119] Creating plugin_oauth_connections'; END $$;
CREATE TABLE IF NOT EXISTS plugin_oauth_connections (
id VARCHAR(36) PRIMARY KEY,
plugin_id VARCHAR(128) NOT NULL,
tenant_id BIGINT NOT NULL,
scope VARCHAR(16) NOT NULL,
contribution VARCHAR(160) NOT NULL DEFAULT '',
field VARCHAR(256) NOT NULL,
token TEXT NOT NULL,
expires_at TIMESTAMPTZ,
created_by VARCHAR(64) NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_plugin_oauth_connections_owner ON plugin_oauth_connections (plugin_id, tenant_id);
COMMENT ON TABLE plugin_oauth_connections IS 'OAuth tokens behind plugin form fields (x-oauth), sealed.';
+67
View File
@@ -151,6 +151,69 @@ WeKnora answers 404 unless all of these hold:
Bodies are limited to 1 MB, with 20 calls a second per URL.
## Dynamic choices and OAuth
Two schema keywords let a form ask the plugin while someone fills it in.
**`x-options`** loads a field's choices from the plugin, for lists that
depend on the account (projects, spaces, channels). It works on a string or
an array of strings:
```yaml
project:
type: string
x-options: { name: projects, dependsOn: [site, token], search: true }
```
```go
p.Options("projects", func(ctx context.Context, call *pluginsdk.Call, in pluginapi.OptionsInput) ([]pluginapi.Option, error) {
// call.Config holds what the form holds now, secrets included.
if call.Config.Tenant["token"] == nil {
return nil, pluginapi.InvalidConfig("enter the token first", map[string]string{"token": "required"})
}
return []pluginapi.Option{{Value: "p1", Label: "Project " + in.Query}}, nil
})
```
- `in.Scope` is `system`, `tenant` or `instance`; the values are in the
matching part of `call.Config`.
- Secrets the form shows redacted are filled in from the stored
configuration.
- The form reloads the list when a `dependsOn` field changes. With
`search`, `in.Query` is what the user typed.
- WeKnora does not check a saved value against the list; the plugin
should.
**`x-oauth`** turns a string field into an account connection. WeKnora runs
the OAuth 2.0 authorization code flow and keeps the tokens:
```yaml
config:
system: schemas/system.yaml # client_id, client_secret (x-secret)
# schemas/tenant.yaml
account:
type: string
x-oauth:
authorizeUrl: https://auth.example.com/oauth/authorize
tokenUrl: https://auth.example.com/oauth/token
scopes: [read]
clientId: ${system.client_id}
clientSecret: ${system.client_secret}
pkce: true
params: { audience: api.example.com } # extra authorize parameters
```
- The form shows a connect button that opens the provider's consent page.
- The field stores `oauth:<connection>`. At call time the plugin receives a
fresh access token in its place; WeKnora refreshes it when it is about to
expire. A connection that can no longer be refreshed arrives as `""`, so
answer `unauthorized` then.
- The platform registers the OAuth app with the provider, with the
redirect URI `<WeKnora address>/api/v1/plugin-oauth/callback`, and enters
its credentials in the plugin's platform configuration.
- `tokenUrl` must be https. WeKnora calls it itself, so it needs no
`permissions.egress` entry.
## Calling back into WeKnora (Host API)
A plugin that declares `permissions.hostApi` gets a short-lived token with
@@ -201,6 +264,10 @@ A few rules the manifest and host enforce:
- **Outbound traffic.** It goes through the host's egress proxy, which
forwards only to the hosts in `permissions.egress` and never to private
addresses.
- **Resources.** `runtime.resources: { cpu: "500m", memory: 512Mi }`
caps the process (Linux). Memory is always capped. CPU is capped only
when the platform delegates a cgroup to WeKnora
(`WEKNORA_PLUGIN_CGROUP`).
- **IDs.** Contribution IDs are qualified with the plugin ID
(`acme.notes/notes`); for connectors and web search that must stay within
50 characters.
+11
View File
@@ -159,6 +159,17 @@ func Run(ctx context.Context, t Target) Report {
return protocolAnswer(t.Client.Call(ctx, pluginapi.EventsPath, envelope(), ev, nil))
})
}
for _, name := range m.Contributes["options"] {
check("options/"+name+" answers", func(ctx context.Context) error {
var out pluginapi.OptionsOutput
in := pluginapi.OptionsInput{Field: "field", Scope: pluginapi.OptionsScopeTenant}
err := t.Client.Call(ctx, pluginapi.OptionsPath(name), envelope(), in, &out)
if err == nil && out.Options == nil {
return fmt.Errorf("output has no options array")
}
return protocolAnswer(err)
})
}
for _, id := range m.Contributes["webhooks"] {
check("webhooks/"+id+" answers", func(ctx context.Context) error {
var out pluginapi.WebhookResponse
+1 -1
View File
@@ -25,7 +25,7 @@ func TestOpenAPIMatchesRoutes(t *testing.T) {
if err != nil {
t.Fatal(err)
}
for _, f := range []string{"websearch.go", "connector.go", "parser.go", "ui.go", "events.go"} {
for _, f := range []string{"websearch.go", "connector.go", "parser.go", "ui.go", "events.go", "options.go"} {
b, err := os.ReadFile(f)
if err != nil {
t.Fatal(err)
+42
View File
@@ -0,0 +1,42 @@
package pluginsdk
import (
"context"
"encoding/json"
"net/http"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// OptionsHandler lists a field's choices. call.Config holds what the form
// holds now; decode it with call.DecodeInstance for instance forms.
type OptionsHandler func(ctx context.Context, call *Call, in pluginapi.OptionsInput) ([]pluginapi.Option, error)
// Options registers the choices behind x-options {name: name}. Return
// pluginapi.InvalidConfig when the form lacks what the list needs (a token
// not entered yet): the form shows the message by the field.
func (p *Plugin) Options(name string, h OptionsHandler) { p.options[name] = h }
func (p *Plugin) routeOptions(mux *http.ServeMux) {
mux.HandleFunc("POST /v1/options/{name}", func(w http.ResponseWriter, r *http.Request) {
h, ok := p.options[r.PathValue("name")]
if !ok {
writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no options %q", r.PathValue("name")))
return
}
p.unary(func(ctx context.Context, call *Call, raw json.RawMessage) (any, error) {
in, err := decodeInput[pluginapi.OptionsInput](raw)
if err != nil {
return nil, err
}
opts, err := h(ctx, call, in)
if err != nil {
return nil, err
}
if opts == nil {
opts = []pluginapi.Option{}
}
return pluginapi.OptionsOutput{Options: opts}, nil
})(w, r)
})
}
+4
View File
@@ -66,6 +66,7 @@ type Plugin struct {
ui UIHandler
events EventHandler
webhooks map[string]WebhookHandler
options map[string]OptionsHandler
validate ConfigValidator
logger *slog.Logger
// ShutdownTimeout bounds how long Serve waits for calls in flight after
@@ -81,6 +82,7 @@ func New(info Info) *Plugin {
connectors: map[string]Connector{},
parsers: map[string]Parser{},
webhooks: map[string]WebhookHandler{},
options: map[string]OptionsHandler{},
logger: slog.New(slog.NewTextHandler(os.Stderr, nil)),
ShutdownTimeout: 60 * time.Second,
}
@@ -108,6 +110,7 @@ func (p *Plugin) Manifest() pluginapi.Manifest {
add("connectors", keys(p.connectors))
add("parsers", keys(p.parsers))
add("webhooks", keys(p.webhooks))
add("options", keys(p.options))
if p.events != nil {
m.Contributes["events"] = []string{"handler"}
}
@@ -151,6 +154,7 @@ func (p *Plugin) Handler() http.Handler {
p.routeParsers(mux)
p.routeUI(mux)
p.routeEvents(mux)
p.routeOptions(mux)
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no endpoint %s %s", r.Method, r.URL.Path))
})
+28
View File
@@ -344,3 +344,31 @@ func TestEventsAndWebhooks(t *testing.T) {
t.Fatalf("unknown webhook = %v", err)
}
}
func TestOptions(t *testing.T) {
ctx := context.Background()
p := New(Info{ID: "acme.opts", Version: "1.0.0"})
p.Options("projects", func(_ context.Context, call *Call, in pluginapi.OptionsInput) ([]pluginapi.Option, error) {
if call.Config.Tenant["token"] == nil {
return nil, pluginapi.InvalidConfig("enter a token first", map[string]string{"token": "required"})
}
return []pluginapi.Option{{Value: "p1", Label: "Project " + in.Query}}, nil
})
if m := p.Manifest(); m.Contributes["options"][0] != "projects" {
t.Fatalf("manifest = %+v", m)
}
srv := httptest.NewServer(p.Handler())
defer srv.Close()
c := client.New(srv.URL, nil, nil)
in := pluginapi.OptionsInput{Field: "project", Scope: "tenant", Query: "x"}
var out pluginapi.OptionsOutput
err := c.Call(ctx, pluginapi.OptionsPath("projects"), pluginapi.Envelope{}, in, &out)
if !isCode(err, pluginapi.CodeInvalidConfig) {
t.Fatalf("without a token = %v", err)
}
env := pluginapi.Envelope{Config: pluginapi.Config{Tenant: map[string]any{"token": "t"}}}
err = c.Call(ctx, pluginapi.OptionsPath("projects"), env, in, &out)
if err != nil || out.Options[0].Label != "Project x" {
t.Fatalf("options = %+v, %v", out, err)
}
}
+49
View File
@@ -218,6 +218,34 @@ paths:
required: [output]
properties: { output: { $ref: "#/components/schemas/UIResponse" } }
default: { $ref: "#/components/responses/Error" }
/v1/options/{name}:
post:
summary: List the choices of a form field (x-options)
description: |
The envelope config holds what the form holds now, stored secrets
filled in for fields the user did not change. Answer invalid_config
when the form lacks what the list needs. The manifest lists the
names under contributes.options.
parameters:
- { name: name, in: path, required: true, schema: { type: string } }
requestBody:
required: true
content:
application/json:
schema:
allOf:
- $ref: "#/components/schemas/Envelope"
- properties: { input: { $ref: "#/components/schemas/OptionsInput" } }
responses:
"200":
description: The choices
content:
application/json:
schema:
type: object
required: [output]
properties: { output: { $ref: "#/components/schemas/OptionsOutput" } }
default: { $ref: "#/components/responses/Error" }
/v1/events:
post:
summary: Receive an event the plugin subscribed to (permissions.events)
@@ -429,6 +457,27 @@ components:
mimeType: { type: string }
data: { type: string, contentEncoding: base64 }
metadata: { type: object, additionalProperties: { type: string } }
OptionsInput:
type: object
required: [field, scope]
properties:
field: { type: string, description: Dotted path of the field }
scope: { enum: [system, tenant, instance] }
contribution: { type: string, description: '"<point>/<id>" when scope is instance' }
query: { type: string }
OptionsOutput:
type: object
required: [options]
properties:
options:
type: array
items:
type: object
required: [value, label]
properties:
value: {}
label: { type: string }
description: { type: string }
EventDelivery:
type: object
required: [id, type, occurredAt, attempt]
+40
View File
@@ -0,0 +1,40 @@
package pluginapi
// OptionsPath answers the choices of a form field whose schema names
// x-options {name}: projects of the account the form is being filled for,
// channels of a workspace, and so on. The envelope's config carries what
// the form holds right now (stored secrets filled in for fields the user did
// not change), in the scope the form edits.
func OptionsPath(name string) string { return "/v1/options/" + name }
// Option scopes: which configuration the form edits.
const (
OptionsScopeSystem = "system"
OptionsScopeTenant = "tenant"
OptionsScopeInstance = "instance"
)
// OptionsInput asks for one field's choices.
type OptionsInput struct {
// Field is the dotted path of the field ("settings.project").
Field string `json:"field"`
// Scope is system, tenant or instance.
Scope string `json:"scope"`
// Contribution is "<point>/<id>" of the instance's contribution
// ("connectors/jira") when Scope is instance.
Contribution string `json:"contribution,omitempty"`
// Query is what the user typed, for fields that search (x-options.search).
Query string `json:"query,omitempty"`
}
// Option is one choice.
type Option struct {
Value any `json:"value"`
Label string `json:"label"`
Description string `json:"description,omitempty"`
}
// OptionsOutput lists the choices, in display order.
type OptionsOutput struct {
Options []Option `json:"options"`
}
+17
View File
@@ -89,6 +89,23 @@ def notes(call, req: WebhookRequest): # req.headers, req.body, req.json()
return WebhookResponse(status=204)
```
**Dynamic choices.** A field with `x-options: {name: projects}` in a
config schema gets its choices from the plugin. Return `Option`s or
`(value, label)` pairs:
```python
@plugin.options("projects")
def projects(call, inp: OptionsInput):
# inp.scope, inp.field, inp.query; the form's values are in call.tenant / call.instance
if not call.tenant.get("token"):
raise invalid_config("enter the token first", {"token": "required"})
return [("p1", "Project one"), ("p2", "Project two")]
```
**OAuth.** An `x-oauth` field needs no code: WeKnora runs the flow, and
the plugin finds a fresh access token where the field is (see the
[Go SDK README](../README.md#dynamic-choices-and-oauth)).
## Errors
Raise `PluginError(ErrorCode.X, "message")` to choose what WeKnora sees:
@@ -32,6 +32,8 @@ from .types import (
EventDelivery,
FetchedItem,
FetchInput,
Option,
OptionsInput,
KVEntry,
KVList,
ParsedImage,
@@ -67,6 +69,8 @@ __all__ = [
"Host",
"KVEntry",
"KVList",
"Option",
"OptionsInput",
"ParseInput",
"ParseOutput",
"ParsedImage",
@@ -32,6 +32,7 @@ from .types import (
SearchInput,
SearchResult,
EventDelivery,
OptionsInput,
UIRequest,
UIResponse,
WebhookRequest,
@@ -167,6 +168,7 @@ ConfigValidator = Callable[[Call], None]
UIHandler = Callable[[Call, UIRequest], Any]
EventHandler = Callable[[Call, EventDelivery], None]
WebhookHandler = Callable[[Call, WebhookRequest], Any]
OptionsHandler = Callable[[Call, OptionsInput], Any]
_ROUTE = re.compile(r"^/v1/(websearch|connectors|parsers)/([^/]+)/([a-z-]+)$")
@@ -186,6 +188,7 @@ class Plugin:
self._ui: Optional[UIHandler] = None
self._events: Optional[EventHandler] = None
self._webhooks: Dict[str, WebhookHandler] = {}
self._options: Dict[str, OptionsHandler] = {}
#: Seconds serve() waits for calls in flight after SIGTERM.
self.shutdown_timeout = 60.0
if logger is None:
@@ -241,6 +244,13 @@ class Plugin:
200. call is the workspace the URL belongs to."""
return self._register(self._webhooks, id, fn)
def options(self, name: str, fn: Optional[OptionsHandler] = None) -> Any:
"""Registers the choices behind x-options {name}: fn(call,
OptionsInput) returns a list of Option (or of (value, label)
pairs). call's config holds what the form holds now. Raise
invalid_config(...) when the form lacks what the list needs."""
return self._register(self._options, name, fn)
def ui(self, fn: UIHandler) -> UIHandler:
"""Registers the handler behind the plugin's pages: fn(call,
UIRequest) returns a UIResponse, or any JSON value for a 200."""
@@ -273,6 +283,7 @@ class Plugin:
("connectors", self._connectors),
("parsers", self._parsers),
("webhooks", self._webhooks),
("options", self._options),
):
if table:
contributes[point] = sorted(table)
@@ -307,6 +318,14 @@ class Plugin:
events = self._events
self._unary(h, body, lambda call, raw: events(call, from_wire(EventDelivery, raw)), empty=True)
return
if method == "POST" and path.startswith("/v1/options/"):
name = path[len("/v1/options/"):]
fn = self._options.get(name)
if fn is None:
h._send_error(PluginError(ErrorCode.NOT_FOUND, f"no options {name!r}"))
else:
self._unary(h, body, lambda call, raw: {"options": _options_output(fn(call, from_wire(OptionsInput, raw)))})
return
if method == "POST" and path.startswith("/v1/webhooks/"):
hook = self._webhooks.get(path[len("/v1/webhooks/"):])
if hook is None:
@@ -474,6 +493,17 @@ def _connector_call(c: Any, action: str, call: Call, raw: Any) -> Any:
return {"ancestors": list(resolve(call, cfg, list(raw.get("resourceIds") or [])) or [])}
def _options_output(out: Any) -> list:
from .types import Option
result = []
for o in out or []:
if isinstance(o, (tuple, list)) and len(o) == 2:
o = Option(value=o[0], label=str(o[1]))
result.append(o)
return result
def _webhook_output(out: Any) -> Any:
if not isinstance(out, WebhookResponse):
import json as _json
@@ -312,3 +312,25 @@ class WebhookResponse:
def __post_init__(self) -> None:
if isinstance(self.body, str):
self.body = self.body.encode()
@dataclass
class OptionsInput:
"""Asks for one form field's choices (x-options). field is its dotted
path; scope is system, tenant or instance; contribution names the
instance's contribution ("connectors/jira"); query is what the user
typed."""
field: str = ""
scope: str = ""
contribution: str = ""
query: str = ""
@dataclass
class Option:
"""One choice of a form field."""
value: Any
label: str
description: str = ""
+7
View File
@@ -85,6 +85,13 @@ def on_event(call, ev):
seen_events.append((ev.id, ev.type, ev.attempt, call.tenant_id, (ev.data or {}).get("knowledgeId")))
@plugin.options("projects")
def projects(call, inp):
if not call.tenant.get("token"):
raise invalid_config("enter a token first", {"token": "required"})
return [("p1", f"Project {inp.query}"), ("p2", "Other")]
@plugin.webhook("inbox")
def inbox(call, req):
if req.headers.get("X-Signature") != "ok":
+8
View File
@@ -81,6 +81,7 @@ class PluginTest(unittest.TestCase):
"connectors": ["notes"],
"parsers": ["upper"],
"webhooks": ["inbox"],
"options": ["projects"],
"ui": ["request"],
"events": ["handler"],
},
@@ -133,6 +134,13 @@ class PluginTest(unittest.TestCase):
status, body = self.c.call("/v1/events", dict(ev, type="knowledge.failed"))
self.assertEqual((status, body["error"]["retryable"]), (503, True))
def test_options(self):
inp = {"field": "project", "scope": "tenant", "query": "x"}
status, body = self.c.call("/v1/options/projects", inp)
self.assertEqual((status, body["error"]["code"]), (400, "invalid_config"))
status, body = self.c.call("/v1/options/projects", inp, {"tenant": {"token": "t"}})
self.assertEqual(body["output"]["options"], [{"value": "p1", "label": "Project x"}, {"value": "p2", "label": "Other"}])
def test_webhooks(self):
body = base64.b64encode(b'{"a": 1}').decode()
req = {"method": "POST", "path": "/issue", "headers": {"X-Signature": "ok"}, "body": body}
+21
View File
@@ -50,6 +50,9 @@
- 出网流量经宿主的出口代理,只放行插件在 `permissions.egress` 中声明的域名,内网地址一律拒绝。
- 宿主检查运行的进程与安装的包一致,健康检查失败时按退避重启,并在插件详情中显示为「异常」。
- Python 插件使用本机的 `python3`(可用 `WEKNORA_PLUGIN_PYTHON` 指定解释器)。Docker app 镜像已带 Python。
- 在 Linux 上,插件进程按 `runtime.resources` 限制资源:
- 内存总是受限。插件没有声明时,按 `WEKNORA_PLUGIN_MEMORY_DEFAULT`(如 `1Gi`)限制,未设置则不限。
- CPU 需要把一个可写的 cgroup v2 目录委托给 WeKnora,并通过 `WEKNORA_PLUGIN_CGROUP` 指定。设置后,每个插件进程进入各自的子 cgroup。
### 独立插件宿主
@@ -126,6 +129,24 @@ Helm 设置 `pluginHost.enabled=true` 即可。使用本地存储(`STORAGE_TYP
- 插件自行用空间配置里的密钥校验调用方。
- 设置 `APP_EXTERNAL_URL` 后,插件还能拿到完整地址,自动向第三方注册。
## 动态选项与账号连接
插件的配置表单可以在填写时向插件取数据:
- **动态选项**:下拉框的选项由插件实时给出,例如用填好的令牌列出项目。
- 修改选项所依赖的字段后,列表会重新加载。
- 编辑已保存的配置时,表单上没改动的密钥由服务端补齐,不会发回浏览器。
- **账号连接**:字段显示为「连接账号」按钮,点击后在弹窗中完成第三方的 OAuth 授权。
- 令牌只保存在 WeKnora 服务端(加密)。每次调用插件时,WeKnora 换上有效的访问令牌,并在过期前自动刷新。
- 连接只属于发起它的空间。
使用账号连接前,系统管理员需要:
1. 在第三方平台注册 OAuth 应用,回调地址填 `<WeKnora 地址>/api/v1/plugin-oauth/callback`。
2. 在「设置 → 插件管理」的插件详情中,把应用的 Client ID 和 Secret 填进平台配置。
回调地址取自 `APP_EXTERNAL_URL`;未设置时,使用浏览器访问 WeKnora 所用的地址。
## 开发插件
- **Go**:[pluginsdk](https://github.com/Tencent/WeKnora/tree/main/pluginsdk),含协议定义、SDK、客户端和一致性测试工具 `weknora-plugin-conformance`。