diff --git a/docs/plugins.md b/docs/plugins.md index 0cdfb13..dce8052 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -381,14 +381,39 @@ plugin.http = { 插件可以在运行时安装、改写**自己的**适配器 —— 这是"带自动轮换的认证"的实现 方式：拿着新 token 重写适配器源码，后续请求自然使用新逻辑（内核重建该适配器 -的池）。这是 Phase 3 的桥接能力，安全规则如下： +的池）。**可直接在插件 `.lua` 里调用**： -- 插件**只能改自己装的**适配器（归属表在 Go 侧记录，不靠约定）。 +```lua +-- 每小时的定时任务：轮换 token 并重写适配器 +function plugin.refresh_token() + local tok = fetch_new_token() -- 插件自己的取 token 逻辑 + local name, err = __llmsproxy_adapters.install( + "myauth", + 'return { name = "myauth", endpoint = "/chat/completions",' .. + ' headers = { ["Authorization"] = "Bearer ' .. tok .. '" } }' + ) + if err then log("error", err) end +end +``` + +| 调用 | 参数 | 返回 | +|---|---|---| +| `__llmsproxy_adapters.install(name, source)` | 适配器名 + Lua 源码 | `(生效名, nil)` 或 `(nil, err)` | +| `__llmsproxy_adapters.update(name, source)` | 同上（只限自己装的） | `(true, nil)` 或 `(nil, err)` | +| `__llmsproxy_adapters.remove(name)` | 适配器名（只限自己装的） | `(true, nil)` 或 `(nil, err)` | + +出错时不抛异常，而是返回 `(nil, err)`——与钩子/HTTP 处理器同一套约定， +一个手滑的插件错误不会带走网关。 + +安全规则（由 Go 侧强制，不是约定）： + +- 插件**只能改自己装的**适配器（归属表在 Go 侧记录，调用身份是插件启动时的 + 固定名字，无法冒充）。 - 插件**不能影子内置适配器**（否则等于篡改发行物）；运维仍可自行覆盖。 - 名字经过路径安全校验（无穿越/绝对路径/空白）。 -> 注：这层桥接目前以 Go API 形式提供并被测试覆盖；插件 `.lua` 直接调用它的 -> 全局绑定尚未接通（见 plan.md 已知限制）。 +> 这条能力配合 §5.4 的定时任务，就是完整的自动轮换：**没有浏览器、没有人 +> 操作时，插件按时间自己把 token 换掉**。 --- diff --git a/e2e/adapter_takeover_test.go b/e2e/adapter_takeover_test.go new file mode 100644 index 0000000..0f7c071 --- /dev/null +++ b/e2e/adapter_takeover_test.go @@ -0,0 +1,286 @@ +package e2e + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// 本文件端到端验证 R2/R5 的核心主张：**对外服务的 wire 格式由适配器决定， +// 内核只调度**；适配器不声明时内核的 OpenAI 兜底仍然可用。 +// +// 与其他测试的分工： +// - internal/gateway/messages_test.go 在**进程内**验证同一件事（起 core+gateway， +// 不打真二进制）。它跑得快，但绕过了一些只在真实进程里发生的事：Lua 适配器的 +// 磁盘 seed、配置加载、systemd 式启动路径、真 HTTP 栈。 +// - 本文件打**真编译出的二进制**，用一个只说 Anthropic Messages 的假上游， +// 证明整条链路（真进程 → 真 Lua 适配器 → 真上游 → 适配器序列化 → 客户端） +// 确实成立，而不是只在进程内装配下成立。 +// +// 这是用户明确问到的"适配器接手对外服务的部分"的验收点。 + +// anthropicMockUpstream 是只懂 Anthropic Messages 协议的假上游。 +// +// 它会对**不符形状的请求主动报错**，而不是宽容接受 —— 宽容的 mock 会让 +// "内核把 OpenAI 形状发给了 Anthropic 上游"这种错误测试通过。 +func anthropicMockUpstream(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Anthropic 的请求路径必须是 /messages + if !strings.HasSuffix(r.URL.Path, "/messages") { + t.Errorf("请求打到了非 Anthropic 路径: %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + body, _ := io.ReadAll(r.Body) + var probe map[string]interface{} + if err := json.Unmarshal(body, &probe); err != nil { + t.Errorf("上游收到非 JSON 请求体: %s", body) + w.WriteHeader(http.StatusBadRequest) + return + } + // Anthropic 形状的硬性要求：messages 是数组；不能有 OpenAI 的 choices 概念 + if _, ok := probe["messages"]; !ok { + t.Errorf("上游请求体缺 messages: %s", body) + } + if _, ok := probe["max_tokens"]; !ok { + t.Errorf("Anthropic 要求 max_tokens，请求体缺此项: %s", body) + } + // 反向断言：OpenAI 兜底字段不该出现在 Anthropic 上游请求里 + for _, k := range []string{"choices", "chat_completions"} { + if _, bad := probe[k]; bad { + t.Errorf("Anthropic 上游请求体混入了 OpenAI 字段 %q: %s", k, body) + } + } + + if stream, _ := probe["stream"].(bool); stream { + w.Header().Set("Content-Type", "text/event-stream") + fl, _ := w.(http.Flusher) + frames := []string{ + `event: message_start` + "\n" + `data: {"type":"message_start","message":{"id":"msg_e2e","type":"message","role":"assistant","model":"claude-e2e","content":[],"stop_reason":null,"usage":{"input_tokens":3,"output_tokens":0}}}`, + `event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, + `event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"pong"}}`, + `event: content_block_stop` + "\n" + `data: {"type":"content_block_stop","index":0}`, + `event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}`, + `event: message_stop` + "\n" + `data: {"type":"message_stop"}`, + } + for _, f := range frames { + fmt.Fprint(w, f+"\n\n") + if fl != nil { + fl.Flush() + } + } + return + } + + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"id":"msg_e2e","type":"message","role":"assistant","model":"claude-e2e",`+ + `"content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn",`+ + `"usage":{"input_tokens":3,"output_tokens":1}}`) + })) + t.Cleanup(srv.Close) + return srv +} + +// writeAnthropicConfig 写一份把源挂到 anthropic 适配器上的配置。 +func writeAnthropicConfig(t *testing.T, dir, listenAddr, upstreamURL string) string { + t.Helper() + cfg := "listen: " + listenAddr + "\n" + + "gateway_keys:\n - sk-e2e-0001\n" + + "default_model: claude-e2e\n" + + "adapter_dir: " + filepath.Join(dir, "adapters") + "\n" + + "runtime_file: " + filepath.Join(dir, "runtime.json") + "\n" + + "sources:\n" + + " - name: anth\n" + + " base_url: " + upstreamURL + "\n" + + " adapter: anthropic\n" + + " api_key: test-key\n" + + " max_concurrent: 32\n" + + " models:\n" + + " - id: claude-e2e\n" + + " kind: chat\n" + + " priority: 100\n" + path := filepath.Join(dir, "config.yaml") + if err := os.WriteFile(path, []byte(cfg), 0o644); err != nil { + t.Fatalf("write config: %v", err) + } + return path +} + +// TestAdapterTakeoverNonStream：非流式，Anthropic 客户端形状进、出。 +func TestAdapterTakeoverNonStream(t *testing.T) { + up := anthropicMockUpstream(t) + dir := t.TempDir() + addr := mustAddr(t) + cfg := writeAnthropicConfig(t, dir, addr, up.URL) + bin := buildBinary(t) + g := startGateway(t, bin, addr, cfg) + + resp, body := g.do("POST", "/v1/messages", + `{"model":"claude-e2e","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`, true) + if resp == nil || resp.StatusCode != 200 { + t.Fatalf("status=%v body=%q", statusOf(resp), body) + } + // 出向必须是 Anthropic Messages 形状，不能是 OpenAI choices 形状 + if strings.Contains(body, `"choices"`) { + t.Errorf("Anthropic 响应里混入了 OpenAI choices 外壳: %s", body) + } + var out map[string]interface{} + if err := json.Unmarshal([]byte(body), &out); err != nil { + t.Fatalf("响应不是 JSON: %v: %s", err, body) + } + if out["type"] != "message" { + t.Errorf(`响应 type = %v, 期望 "message"（Anthropic 形状）: %s`, out["type"], body) + } + // 文本内容必须能取出来 —— 这是"内容确实穿过适配器回来了"的证据 + if !strings.Contains(body, "pong") { + t.Errorf("响应里没有上游返回的文本: %s", body) + } + if _, ok := out["usage"]; !ok { + t.Errorf("Anthropic 响应应带 usage: %s", body) + } +} + +// TestAdapterTakeoverStream：流式也要是 Anthropic SSE 序列， +// 且**不得**出现 OpenAI 的 [DONE] 终止标记（R2 的"不混合"）。 +func TestAdapterTakeoverStream(t *testing.T) { + up := anthropicMockUpstream(t) + dir := t.TempDir() + addr := mustAddr(t) + cfg := writeAnthropicConfig(t, dir, addr, up.URL) + bin := buildBinary(t) + g := startGateway(t, bin, addr, cfg) + + resp, body := g.do("POST", "/v1/messages", + `{"model":"claude-e2e","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"hi"}]}`, true) + if resp == nil || resp.StatusCode != 200 { + t.Fatalf("stream status=%v body=%q", statusOf(resp), body) + } + if !strings.Contains(body, "message_start") { + t.Errorf("Anthropic 流应以 message_start 开头: %s", body) + } + if !strings.Contains(body, "message_stop") { + t.Errorf("Anthropic 流应以 message_stop 收尾: %s", body) + } + if strings.Contains(body, "[DONE]") { + t.Errorf("Anthropic 流里出现了 OpenAI 的 [DONE]: %s", body) + } + if strings.Contains(body, "chat.completion.chunk") { + t.Errorf("流里混入 OpenAI chunk 形状: %s", body) + } + if !strings.Contains(body, "pong") { + t.Errorf("流里没有上游文本: %s", body) + } +} + +// TestAdapterTakeoverStress 是对外服务路径的真实压力测试： +// 并发打真二进制，混合流式/非流式，验证无 5xx、上游命中数正确、 +// 且每个响应都是完整合法的 Anthropic 形状（适配器在并发下不串味）。 +func TestAdapterTakeoverStress(t *testing.T) { + up := anthropicMockUpstream(t) + dir := t.TempDir() + addr := mustAddr(t) + cfg := writeAnthropicConfig(t, dir, addr, up.URL) + bin := buildBinary(t) + g := startGateway(t, bin, addr, cfg) + + // max_concurrent 配成 32，让 96 个请求基本都能真正流过适配器 —— 这个测试 + // 要压的是**并发下序列化是否串味**，不是内核的容量拒绝策略。真的撞上上限 + // 时收到的 429 也是合法背压（见下方断言），不算失败。 + const ( + workers = 12 + perWorker = 8 + ) + + var ( + wg sync.WaitGroup + mu sync.Mutex + ok int + shed int + failed []string + streams int + ) + start := make(chan struct{}) + + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + <-start + for i := 0; i < perWorker; i++ { + isStream := (w+i)%2 == 0 + var payload string + if isStream { + payload = `{"model":"claude-e2e","max_tokens":32,"stream":true,"messages":[{"role":"user","content":"hi"}]}` + } else { + payload = `{"model":"claude-e2e","max_tokens":32,"messages":[{"role":"user","content":"hi"}]}` + } + resp, body := g.do("POST", "/v1/messages", payload, true) + mu.Lock() + switch { + case resp == nil: + failed = append(failed, "nil response") + // 429 = 内核的容量背压（provider busy）。它是**正确**行为， + // 不是缺陷：并发超限时就该如实拒绝，而不是排队或伪造响应。 + // 只要求它是结构良好的错误，不要求 200。 + case resp.StatusCode == http.StatusTooManyRequests: + if !strings.Contains(body, "busy") && !strings.Contains(body, "error") { + failed = append(failed, "429 不是结构化错误: "+body) + } else { + shed++ + } + // 5xx 才是真失败 —— 那是内核或适配器崩了。 + case resp.StatusCode >= 500: + failed = append(failed, fmt.Sprintf("5xx status=%d body=%.120s", resp.StatusCode, body)) + case resp.StatusCode != 200: + failed = append(failed, fmt.Sprintf("status=%d body=%.120s", resp.StatusCode, body)) + case !strings.Contains(body, "pong"): + // 200 但内容不对：说明并发下适配器把响应串味了 + failed = append(failed, "missing upstream text: "+body) + case isStream && !strings.Contains(body, "message_stop"): + failed = append(failed, "stream missing message_stop") + case !isStream && strings.Contains(body, `"choices"`): + failed = append(failed, "openai shell leaked: "+body) + default: + ok++ + if isStream { + streams++ + } + } + mu.Unlock() + } + }(w) + } + close(start) + wg.Wait() + + if len(failed) > 0 { + t.Fatalf("%d/%d 请求失败，前 5 例:\n%s", len(failed), workers*perWorker, + strings.Join(failed[:min(5, len(failed))], "\n")) + } + // 至少大部分请求必须真正流过 200 —— 全被背压拒掉说明并发上限配得太低， + // 那这个压力测试就没压到适配器并发。 + if ok < workers*perWorker/2 { + t.Fatalf("只有 %d/%d 请求真正成功，压力未触及适配器并发路径", + ok, workers*perWorker) + } + t.Logf("成功 %d（其中流式 %d），背压 429 %d —— 无 5xx、无 OpenAI 形状泄漏", + ok, streams, shed) +} + +func min(a, b int) int { + if a < b { + return a + } + return b +} + +var _ = time.Second diff --git a/internal/lua/plugin_adapter_bridge.go b/internal/lua/plugin_adapter_bridge.go index e90b9d9..d31d38c 100644 --- a/internal/lua/plugin_adapter_bridge.go +++ b/internal/lua/plugin_adapter_bridge.go @@ -27,6 +27,8 @@ import ( "fmt" "sort" "sync" + + golua "github.com/aarzilli/golua/lua" ) // AdapterOwner 标识一个适配器由谁安装。 @@ -185,3 +187,76 @@ func (ps *Plugins) PluginAdapters(plugin string) []string { func (ps *Plugins) AdapterOwnerOf(name string) string { return ps.adapterOwner().ownedBy(name) } + +// ---- Lua 绑定（R4：插件在运行时改自己的适配器）---- + +// adapterBridgeGlobal 暴露给插件 Lua 的适配器管理表。 +// +// 用一张表而不是三个顶层全局，是为了少占插件可见的全局命名空间。表用 +// buildJSONTable 的同一套写法（NewTable + PushGoClosure + SetField）， +// 该写法在 json 表上已被长期验证。 +const adapterBridgeGlobal = "__llmsproxy_adapters" + +// registerAdapterBridge 把适配器管理表装进插件的 Lua 状态。 +// +// name 是**该插件自己的启动名**，随每次调用一起交给 Go 侧桥接 —— 归属校验 +// 因此来自启动时的固定身份，而不是从 Lua 状态里现读（那样一个分片或任何持有 +// 已编译脚本的调用者就能冒充别的插件）。 +func registerAdapterBridge(L *golua.State, ps *Plugins, name string) { + L.NewTable() + install := luaInstallAdapter(L, ps, name) + L.PushGoClosure(install) + L.SetField(-2, "install") + update := luaUpdateAdapter(L, ps, name) + L.PushGoClosure(update) + L.SetField(-2, "update") + remove := luaRemoveAdapter(L, ps, name) + L.PushGoClosure(remove) + L.SetField(-2, "remove") + L.SetGlobal(adapterBridgeGlobal) +} + +func luaInstallAdapter(L *golua.State, ps *Plugins, name string) golua.LuaGoFunction { + return func(L *golua.State) int { + adapterName := L.ToString(1) + code := L.ToString(2) + effective, err := ps.InstallAdapter(name, adapterName, code) + if err != nil { + L.PushNil() + L.PushString(err.Error()) + return 2 + } + L.PushString(effective) + L.PushNil() + return 2 + } +} + +func luaUpdateAdapter(L *golua.State, ps *Plugins, name string) golua.LuaGoFunction { + return func(L *golua.State) int { + adapterName := L.ToString(1) + code := L.ToString(2) + if err := ps.UpdateAdapter(name, adapterName, code); err != nil { + L.PushNil() + L.PushString(err.Error()) + return 2 + } + L.PushBoolean(true) + L.PushNil() + return 2 + } +} + +func luaRemoveAdapter(L *golua.State, ps *Plugins, name string) golua.LuaGoFunction { + return func(L *golua.State) int { + adapterName := L.ToString(1) + if err := ps.RemoveAdapter(name, adapterName); err != nil { + L.PushNil() + L.PushString(err.Error()) + return 2 + } + L.PushBoolean(true) + L.PushNil() + return 2 + } +} diff --git a/internal/lua/plugin_adapter_bridge_lua_test.go b/internal/lua/plugin_adapter_bridge_lua_test.go new file mode 100644 index 0000000..bf07092 --- /dev/null +++ b/internal/lua/plugin_adapter_bridge_lua_test.go @@ -0,0 +1,234 @@ +package lua + +import ( + "testing" + "time" +) + +// 本文件验证 R4 的 Lua 侧绑定：**插件自己的 Lua 代码**能在运行时安装/改写/ +// 删除自己的适配器。 +// +// 为什么需要单独一层测试：Go 侧的 InstallAdapter/UpdateAdapter 在 +// plugin_adapter_bridge_test.go 里已直接验证过，但那证明的是"Go API 能工作"。 +// token 轮换的真实形态是插件在定时任务或钩子里**从 Lua 发起**这个调用，所以 +// 绑定本身（__llmsproxy_adapters 表）必须被证明可用 —— 而且它曾经在 cgo 层 +// SIGSEGV，所以"不崩"本身就是一条断言。 + +const bridgeProbe = ` +local plugin = { name = "authtoken", version = "1.0" } +plugin.exports = { "install_via_lua", "update_via_lua", "remove_via_lua", "install_operator_owned" } + +local function src(ver, tok) + return 'return { name = "myauth", version = "' .. ver .. '", endpoint = "/x", token = "' .. tok .. '" }' +end + +function plugin.install_via_lua(args) + local name, err = __llmsproxy_adapters.install("myauth", src("1.0", args.tok)) + return { name = name, err = err } +end + +function plugin.update_via_lua(args) + local ok, err = __llmsproxy_adapters.update("myauth", src("2.0", args.tok)) + return { ok = ok, err = err } +end + +function plugin.remove_via_lua() + local ok, err = __llmsproxy_adapters.remove("myauth") + return { ok = ok, err = err } +end + +-- 试图替换一个非自己拥有的适配器：必须被拒（不能影子内置） +function plugin.install_operator_owned() + local name, err = __llmsproxy_adapters.install("openai", src("9.9", "evil")) + return { name = name, err = err } +end +return plugin +` + +// TestPluginInstallsAdapterFromLua 是 R4 绑定的主断言。 +func TestPluginInstallsAdapterFromLua(t *testing.T) { + _, ps, _ := runtimePlugins(t) + if err := ps.LoadSource("authtoken", bridgeProbe); err != nil { + t.Fatal(err) + } + out, err := ps.Call("authtoken", "install_via_lua", map[string]interface{}{"tok": "T1"}) + if err != nil { + t.Fatalf("Lua 侧 install 调用失败: %v", err) + } + m, ok := out.(map[string]interface{}) + if !ok { + t.Fatalf("返回值不是 table: %#v", out) + } + if m["name"] != "myauth" { + t.Fatalf("install 应返回生效名 myauth, 实得 %#v (err=%#v)", m["name"], m["err"]) + } + // 归属确实被记录（这条规则是"插件只能改自己装的"的基础） + if owner := ps.AdapterOwnerOf("myauth"); owner != "authtoken" { + t.Errorf("归属应为 authtoken, 实得 %q", owner) + } +} + +// TestPluginRotatesTokenViaLua 是 token 轮换的完整形态： +// 同一个插件先装、再改，改后新版本生效。 +func TestPluginRotatesTokenViaLua(t *testing.T) { + _, ps, _ := runtimePlugins(t) + if err := ps.LoadSource("authtoken", bridgeProbe); err != nil { + t.Fatal(err) + } + if _, err := ps.Call("authtoken", "install_via_lua", map[string]interface{}{"tok": "OLD"}); err != nil { + t.Fatalf("install: %v", err) + } + // 轮换 + out, err := ps.Call("authtoken", "update_via_lua", map[string]interface{}{"tok": "NEW"}) + if err != nil { + t.Fatalf("update: %v", err) + } + m := out.(map[string]interface{}) + if m["ok"] != true { + t.Fatalf("Lua 侧 update 未成功: %#v", m) + } + // 新代码必须真的生效 —— 这是"轮换有意义"的定义 + if rev := ps.vm.AdapterRevision("myauth"); rev < 2 { + t.Errorf("适配器修订号应 >=2（已被替换一次），实得 %d", rev) + } +} + +// TestPluginRemoveAdapterFromLua：插件能删掉自己装的。 +func TestPluginRemoveAdapterFromLua(t *testing.T) { + _, ps, _ := runtimePlugins(t) + if err := ps.LoadSource("authtoken", bridgeProbe); err != nil { + t.Fatal(err) + } + if _, err := ps.Call("authtoken", "install_via_lua", map[string]interface{}{"tok": "T"}); err != nil { + t.Fatal(err) + } + out, err := ps.Call("authtoken", "remove_via_lua", map[string]interface{}{}) + if err != nil { + t.Fatalf("remove: %v", err) + } + if out.(map[string]interface{})["ok"] != true { + t.Fatalf("Lua 侧 remove 未成功: %#v", out) + } + if owner := ps.AdapterOwnerOf("myauth"); owner != "" { + t.Errorf("删除后归属应清空, 实得 %q", owner) + } +} + +// TestLuaBridgeCannotShadowBuiltin：从 Lua 侧同样不能影子内置适配器 —— +// 否则绑定就成了绕过 Go 侧校验的后门。 +func TestLuaBridgeCannotShadowBuiltin(t *testing.T) { + _, ps, _ := runtimePlugins(t) + if err := ps.LoadSource("authtoken", bridgeProbe); err != nil { + t.Fatal(err) + } + out, err := ps.Call("authtoken", "install_operator_owned", map[string]interface{}{}) + if err != nil { + t.Fatalf("调用本身不应报错（错误应作为返回值）: %v", err) + } + m := out.(map[string]interface{}) + if m["name"] != nil { + t.Fatalf("影子内置 openai 必须被拒，却返回了 name=%#v", m["name"]) + } + if m["err"] == nil || m["err"] == "" { + t.Fatal("影子内置应带回错误信息") + } + t.Logf("拒绝信息: %v", m["err"]) +} + +// TestScheduledTaskRotatesTokenEndToEnd 是 R4 的闭环验收。 +// +// 它把三个特性串成一条链，任何一个环节坏掉都会让它失败： +// 1. 内核 ticker 驱动定时任务（无外部触发） +// 2. 任务与钩子跑在同一个 VM（所以任务的 install 与 state 写入对钩子可见） +// 3. Lua 侧适配器桥接可用（__llmsproxy_adapters.install） +// +// 这正是用户描述的 token 轮换形态：插件自己按时间拿到新 token，重写适配器， +// 后续请求自然使用新逻辑 —— 全程没有任何浏览器或人工操作。 +func TestScheduledTaskRotatesTokenEndToEnd(t *testing.T) { + shortTick(t) + _, ps, _ := runtimePlugins(t) + code := ` +local plugin = { name = "sched_rot", version = "1.0" } +plugin.state = { token = "OLD", rotates = 0 } +plugin.schedule = { { fn = "rotate", every = 1 } } +function plugin.rotate() + plugin.state.rotates = plugin.state.rotates + 1 + local tok = "T" .. tostring(plugin.state.rotates) + local src = 'return { name = "rotauth", version = "1.0", endpoint = "/x", token = "' .. tok .. '" }' + plugin.state.last_adapter, plugin.state.rotate_err = __llmsproxy_adapters.install("rotauth", src) + plugin.state.token = tok +end +plugin.hooks = { routed = "observe" } +function plugin.observe(payload) + payload.token = plugin.state.token + payload.rotates = plugin.state.rotates + payload.adapter = plugin.state.last_adapter + return payload +end +return plugin +` + if err := ps.LoadSource("sched_rot", code); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(4 * time.Second) + for time.Now().Before(deadline) { + out := ps.Fire(StageRouted, map[string]interface{}{}) + if n, ok := out["rotates"].(float64); ok && n >= 1 { + t.Logf("任务轮换了 %v 次; 钩子看到 token=%v adapter=%v", n, out["token"], out["adapter"]) + if out["adapter"] != "rotauth" { + t.Fatalf("任务里 install 未生效: adapter=%v", out["adapter"]) + } + return + } + time.Sleep(30 * time.Millisecond) + } + t.Fatal("定时任务未完成 token 轮换") +} + +// TestShardVMHasNoAdapterBridge 锁定一条边界：分片 VM（ShardCall 用的私有 VM） +// **不能**拿到适配器管理桥接。 +// +// 理由：分片是文档化的「只读、可丢弃」工作单元，而 install/update/remove 是 +// **全局**副作用（改的是共享的适配器注册表，不是本 VM 的状态）。若把桥接给 +// 分片，一个被声明为只读的按需调用就能悄悄改写所有请求共用的适配器 —— +// 正是分片契约要防的那类不可见变更。 +// +// 断言方式是：分片里的插件代码检查 __llmsproxy_adapters 是否为 nil，返回结果。 +func TestShardVMHasNoAdapterBridge(t *testing.T) { + _, ps, _ := runtimePlugins(t) + code := ` +local plugin = { name = "sharded", version = "1.0" } +plugin.exports = { "probe" } +-- 只读探测：分片里这个全局必须是 nil +function plugin.probe(args) + return { has_bridge = __llmsproxy_adapters ~= nil } +end +return plugin +` + if err := ps.LoadSource("sharded", code); err != nil { + t.Fatal(err) + } + // 主 VM 上应当有桥接（对照） + mainOut, err := ps.Call("sharded", "probe", map[string]interface{}{}) + if err != nil { + t.Fatalf("主 VM 调用失败: %v", err) + } + if m, ok := mainOut.(map[string]interface{}); !ok || m["has_bridge"] != true { + t.Errorf("主 VM 上应有 __llmsproxy_adapters, 实得 %#v", mainOut) + } + + // 分片 VM 上不应有 + outs, err := ps.ShardCall("sharded", "probe", []map[string]interface{}{{}, {}}) + if err != nil { + t.Fatalf("ShardCall: %v", err) + } + for i, o := range outs { + m, ok := o.(map[string]interface{}) + if !ok { + t.Fatalf("分片 %d 返回非 table: %#v", i, o) + } + if m["has_bridge"] == true { + t.Errorf("分片 %d 不应拿到适配器桥接（分片是只读可丢弃的）", i) + } + } +} diff --git a/internal/lua/plugin_shard.go b/internal/lua/plugin_shard.go index b39bc5f..fe912fd 100644 --- a/internal/lua/plugin_shard.go +++ b/internal/lua/plugin_shard.go @@ -164,7 +164,14 @@ func (sp *shardPool) acquire(code, name string) (*worker, func(), error) { sp.live++ sp.mu.Unlock() - w, err := bootPluginState(code, name, sp.ps.outboundAllowlist()) + // ps is deliberately nil: a shard VM must NOT get the adapter-management + // bridge. A shard is documented as READ-ONLY, disposable work, and adapter + // install/update/remove is a GLOBAL side effect (it mutates the shared + // adapter registry, not this VM's own state). Handing it to a throwaway VM + // would let a "read-only" on-demand call silently rewrite the adapters every + // other request uses — exactly the invisible-mutation class the shard + // contract exists to prevent. + w, err := bootPluginState(code, name, sp.ps.outboundAllowlist(), nil) if err != nil { sp.mu.Lock() sp.live-- diff --git a/internal/lua/plugins.go b/internal/lua/plugins.go index 556ea7b..b248c0f 100644 --- a/internal/lua/plugins.go +++ b/internal/lua/plugins.go @@ -593,7 +593,7 @@ func (ps *Plugins) SeedBundled() error { // bootPluginState compiles one plugin into a fresh Lua state and stores its // returned table under pluginGlobal. It is the plugin counterpart of // adapterPool.boot, minus the pooling: a plugin keeps exactly one state. -func bootPluginState(code, name string, allow *httpAllowlist) (*worker, error) { +func bootPluginState(code, name string, allow *httpAllowlist, ps *Plugins) (*worker, error) { L := golua.NewState() L.OpenLibs() setupGlobals(L) @@ -604,6 +604,12 @@ func bootPluginState(code, name string, allow *httpAllowlist) (*worker, error) { if allow != nil { registerFn(L, httpGlobal, luaHTTPFn(L, allow)) } + // R4: plugins — and only plugins — can install/update/remove their own + // adapters at runtime. Scoped here (not in setupGlobals) for the same reason + // as __llmsproxy_http: adapters share setupGlobals and must not get it. + if ps != nil { + registerAdapterBridge(L, ps, name) + } if err := L.DoString(code); err != nil { L.Close() return nil, fmt.Errorf("compile plugin %s: %w", name, err) @@ -658,7 +664,7 @@ func (ps *Plugins) LoadSource(name, code string) error { // One state, held for the plugin's lifetime (see Plugin.state): a hook that // accumulates into plugin.state would otherwise split its totals across // whichever worker it happened to run on. - w, err := bootPluginState(code, name, ps.outboundAllowlist()) + w, err := bootPluginState(code, name, ps.outboundAllowlist(), ps) if err != nil { p.LoadError = err.Error() ps.append(p)