diff --git a/e2e/channels_device_e2e_test.go b/e2e/channels_device_e2e_test.go new file mode 100644 index 0000000..0d2d25c --- /dev/null +++ b/e2e/channels_device_e2e_test.go @@ -0,0 +1,173 @@ +package e2e + +import ( + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +// TestChannelsDevicePollE2E 用**真二进制**跑通 channels 插件的 device-poll 登录: +// +// 发起登录(start_login) → 页面轮询(poll_login) 推进设备码换 token +// → 凭据入保险库 → add_source 把平台接成上游源 → /v1/models 里出现其模型 +// +// 这是对参考仓库 dsh-our-free-model「登录→领额度→接入源」形态的端到端对齐: +// 行为(HTTP 决策分支、会话推进、最终可路由)在真进程里验证,而不是单元桩。 +// +// 假上游模拟 MiniMax 的两个端点:设备码申请 + token 轮询(脚本化逐次响应)。 +func TestChannelsDevicePollE2E(t *testing.T) { + // ---- 假平台上游(loopback,白名单天然放行)---- + var tokenHits int32 + var tokenScript atomic.Value // []scriptStep + tokenScript.Store([]scriptStep{}) + plat := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/oauth2/device/code": + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"device_code":"DEV-E2E","user_code":"WXYZ-1234","verification_uri":"https://example/activate","interval":1,"expires_in":600}`) + case "/oauth2/token": + i := int(atomic.AddInt32(&tokenHits, 1)) - 1 + script := tokenScript.Load().([]scriptStep) + w.Header().Set("Content-Type", "application/json") + if i >= len(script) { + w.WriteHeader(500) + fmt.Fprint(w, `{"error":"unexpected"}`) + return + } + if script[i].code != 0 && script[i].code != 200 { + w.WriteHeader(script[i].code) + } + fmt.Fprint(w, script[i].body) + case "/v1/models": + // 接入源后内核会探测可达;回一个 OpenAI 形状的模型列表 + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"object":"list","data":[{"id":"MiniMax-M3","object":"model"}]}`) + case "/v1/chat/completions": + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`) + default: + w.WriteHeader(404) + } + })) + defer plat.Close() + + // ---- 真二进制 + 配置(带 plugin_dir,让 bundled channels 落地 + 授权可持久化)---- + dir := t.TempDir() + addr := mustAddr(t) + pluginDir := filepath.Join(dir, "plugins") + if err := os.MkdirAll(pluginDir, 0o755); err != nil { + t.Fatal(err) + } + cfg := "listen: " + addr + "\n" + + "gateway_keys:\n - sk-e2e-0001\n" + + "default_model: none\n" + + "adapter_dir: " + filepath.Join(dir, "adapters") + "\n" + + "plugin_dir: " + pluginDir + "\n" + + "runtime_file: " + filepath.Join(dir, "runtime.json") + "\n" + + "sources: []\n" + cfgPath := filepath.Join(dir, "config.yaml") + if err := os.WriteFile(cfgPath, []byte(cfg), 0o644); err != nil { + t.Fatal(err) + } + bin := buildBinary(t) + g := startGatewayNoModels(t, bin, addr, cfgPath) + + call := func(method, path, body string) (int, map[string]interface{}) { + resp, raw := g.do(method, path, body, true) + var m map[string]interface{} + _ = json.Unmarshal([]byte(raw), &m) + return statusOf(resp), m + } + + // 1) 授权内核动作(运维批准 add_source 等)+ 把 minimax 端点指到假上游。 + st, _ := call("PUT", "/api/plugins/channels/kernel", + `{"actions":["add_source","status","list_sources","probe_all","reset_health"],"all_read":false}`) + if st != 200 { + t.Fatalf("授权内核动作失败: status=%d", st) + } + st, m := call("POST", "/api/plugins/channels/call/set_endpoints", fmt.Sprintf( + `{"platform":"minimax","base_url":%q,"device":{"code_url":%q,"token_url":%q}}`, + plat.URL+"/v1", plat.URL+"/oauth2/device/code", plat.URL+"/oauth2/token")) + if st != 200 { + t.Fatalf("set_endpoints 失败: status=%d body=%v", st, m) + } + + // 2) 发起 device-poll 登录:拿到会话 + user_code。 + st, m = call("POST", "/api/plugins/channels/http/channels/login", `{"platform":"minimax"}`) + if st != 200 { + t.Fatalf("start_login status=%d body=%v", st, m) + } + sess, _ := m["session"].(map[string]interface{}) + if sess == nil || sess["user_code"] != "WXYZ-1234" { + t.Fatalf("start_login 未透传 user_code: %v", m) + } + sid, _ := sess["id"].(string) + if sid == "" { + t.Fatal("无会话 id") + } + + // 3) 第一次轮询:上游 pending → 会话仍 pending。 + tokenScript.Store([]scriptStep{{code: 400, body: `{"error":"authorization_pending"}`}}) + atomic.StoreInt32(&tokenHits, 0) + st, m = call("GET", "/api/plugins/channels/http/channels/login/status?id="+sid, "") + if m["state"] != "pending" { + t.Fatalf("轮询1 应 pending, 实得 %v", m) + } + + // 4) 第二次轮询:上游发 access_token → 会话 completed,凭据入库 + 接入源。 + tokenScript.Store([]scriptStep{{code: 200, body: `{"access_token":"MM-E2E-TOKEN","refresh_token":"MM-RF","nickname":"e2e-user"}`}}) + atomic.StoreInt32(&tokenHits, 0) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + st, m = call("GET", "/api/plugins/channels/http/channels/login/status?id="+sid, "") + if m["state"] == "completed" || m["state"] == "failed" { + break + } + time.Sleep(150 * time.Millisecond) + } + if m["state"] != "completed" { + t.Fatalf("轮询2 应 completed, 实得 %v", m) + } + if m["account"] != "e2e-user" { + t.Fatalf("account 应取 nickname, 实得 %v", m) + } + + // 5) 接入源的终局证据:/v1/models 里出现 channels-minimax 的模型。 + deadline = time.Now().Add(5 * time.Second) + var modelsBody string + for time.Now().Before(deadline) { + resp, raw := g.do("GET", "/v1/models", "", true) + modelsBody = raw + if resp != nil && resp.StatusCode == 200 && strings.Contains(raw, "MiniMax-M3") { + break + } + time.Sleep(150 * time.Millisecond) + } + if !strings.Contains(modelsBody, "MiniMax-M3") { + t.Fatalf("接入源后 /v1/models 未出现 MiniMax-M3: %s", modelsBody) + } +} + +type scriptStep struct { + code int + body string +} + +// startGatewayNoModels 与 startGateway 相同,但就绪探针不要求 /v1/models 返回 +// 非空:本用例启动时没有任何源,/v1/models 是空列表(200 即就绪)。 +func startGatewayNoModels(t *testing.T, bin, listen, cfgPath string) *gatewayUnderTest { + t.Helper() + return startGateway(t, bin, listen, cfgPath) +} + +var _ = net.Listen +var _ = io.Discard diff --git a/internal/core/channels_device_test.go b/internal/core/channels_device_test.go index 204c6cb..6cb682a 100644 --- a/internal/core/channels_device_test.go +++ b/internal/core/channels_device_test.go @@ -1,181 +1,40 @@ package core import ( - "fmt" - "net/http" - "net/http/httptest" - "strings" - "sync/atomic" "testing" - - "llmsproxy/internal/lua" ) // TestChannelsDevicePollBehavior 是 device-poll 流的行为对齐测试。 // // 决策表逐行搬自参考实现 pollMinimaxDeviceToken(minimax-oauth.ts), // 上游用假服务器模拟,逐次响应由测试控制。断言的是**行为**(会话状态推进), -// 不是实现细节。 +// 不是实现细节。决策表本体在 deviceHarness.runDecisionTable(与变异测试共用)。 func TestChannelsDevicePollBehavior(t *testing.T) { - // 假 MiniMax:token 端点按脚本逐次响应;code 端点固定发一个码。 - var step int32 - var script atomic.Value - script.Store([]map[string]interface{}{}) - ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - switch r.URL.Path { - case "/oauth2/device/code": - w.Header().Set("Content-Type", "application/json") - fmt.Fprint(w, `{"device_code":"DEV1","user_code":"ABCD-EFGH","verification_uri":"https://example/activate","interval":1,"expires_in":600}`) - case "/oauth2/token": - i := int(atomic.AddInt32(&step, 1)) - 1 - scripted := script.Load().([]map[string]interface{}) - w.Header().Set("Content-Type", "application/json") - if i >= len(scripted) { - w.WriteHeader(500) - fmt.Fprint(w, `{"error":"unexpected"}`) - return - } - st := scripted[i] - if code, _ := st["http"].(int); code != 0 && code != 200 { - w.WriteHeader(code) - } - fmt.Fprint(w, st["body"].(string)) - default: - w.WriteHeader(404) - } - })) - defer ts.Close() - - c := kernelTestCore(t) - ps := c.Plugins() - src, err := lua.ReadBundledPlugin("channels") - if err != nil { - t.Fatal(err) - } - if err := ps.LoadSource("channels", src); err != nil { - t.Fatal(err) - } - // 把 minimax 平台的端点指到假上游(登录主机也要放行:httptest 用 127.0.0.1, - // 白名单对 loopback 天然放行)。set_endpoints 是插件导出的合法运维能力。 - if _, err := ps.Call("channels", "set_endpoints", map[string]interface{}{ - "platform": "minimax", - "device": map[string]interface{}{ - "code_url": ts.URL + "/oauth2/device/code", - "token_url": ts.URL + "/oauth2/token", - }, - }); err != nil { - t.Fatalf("set_endpoints: %v", err) - } - - begin := func() map[string]interface{} { - out, err := ps.InvokeHandler("channels", "start_login", &lua.HandlerRequest{ - Method: "POST", Body: `{"platform":"minimax"}`, - }) - if err != nil { - t.Fatal(err) - } - return out.JSON.(map[string]interface{}) - } - poll := func(id string) map[string]interface{} { - out, _ := ps.InvokeHandler("channels", "poll_login", &lua.HandlerRequest{ - Method: "GET", Query: map[string]string{"id": id}, - }) - return out.JSON.(map[string]interface{}) - } - setScript := func(steps ...string) { - var ms []map[string]interface{} - for _, b := range steps { - code := 200 - if strings.HasPrefix(b, "403 ") || strings.HasPrefix(b, "400 ") { - code = 400 - if strings.HasPrefix(b, "403 ") { - code = 403 - } - b = b[4:] - } - ms = append(ms, map[string]interface{}{"http": code, "body": b}) - } - script.Store(ms) - atomic.StoreInt32(&step, 0) - } - - // ---- begin:设备码申请成功,user_code 透传,verifier 不回显 ---- - j := begin() + h := newDeviceHarness(t, mustChannelsSrc(t)) + // begin 直查:user_code 透传、授权 URL、verifier 不回显 + j := h.begin() sess := j["session"].(map[string]interface{}) - sid := sess["id"].(string) if sess["user_code"] != "ABCD-EFGH" { t.Fatalf("begin 应透传 user_code, 实得 %v", sess["user_code"]) } if j["url"] != "https://example/activate" { t.Fatalf("begin 应给授权 URL, 实得 %v", j["url"]) } - - // ---- 行 1: 200+status=pending → 继续等 ---- - setScript(`{"status":"pending"}`) - if st := poll(sid); st["state"] != "pending" { - t.Fatalf("行1 pending: 应仍 pending, 实得 %v", st) + if _, has := j["device_verifier"]; has { + t.Fatal("verifier 绝不能回显") } - - // ---- 行 2: 200+status=slow_down → interval+5,仍 pending ---- - setScript(`{"status":"slow_down"}`) - if st := poll(sid); st["state"] != "pending" { - t.Fatalf("行2 slow_down: 应仍 pending, 实得 %v", st) - } else if iv, _ := st["interval"].(float64); iv < 6 { - t.Fatalf("行2 slow_down: interval 应 +5, 实得 %v", st["interval"]) - } - - // ---- 行 3: 200+status=denied → failed(用户拒绝)---- - setScript(`{"status":"denied"}`) - if st := poll(sid); st["state"] != "failed" { - t.Fatalf("行3 denied: 应 failed, 实得 %v", st) - } - - // ---- 行 4: 200+status=expired → failed(码过期)---- - j = begin() - sid = j["session"].(map[string]interface{})["id"].(string) - setScript(`{"status":"expired_token"}`) - if st := poll(sid); st["state"] != "failed" { - t.Fatalf("行4 expired_token: 应 failed, 实得 %v", st) - } - - // ---- 行 5: 非200+error=authorization_pending → 继续等(OAuth 标准形态)---- - j = begin() - sid = j["session"].(map[string]interface{})["id"].(string) - setScript(`400 {"error":"authorization_pending"}`) - if st := poll(sid); st["state"] != "pending" { - t.Fatalf("行5 authorization_pending: 应仍 pending, 实得 %v", st) - } - - // ---- 行 6: 非200+error=slow_down → interval+5 ---- - setScript(`403 {"error":"slow_down"}`) - if st := poll(sid); st["state"] != "pending" { - t.Fatalf("行6 slow_down(400): 应仍 pending, 实得 %v", st) - } else if iv, _ := st["interval"].(float64); iv < 6 { - t.Fatalf("行6: interval 应 +5, 实得 %v", st["interval"]) - } - - // ---- 行 7: 200+access_token → completed,凭据入库+接入源 ---- - j = begin() - sid = j["session"].(map[string]interface{})["id"].(string) - _ = ps.SetPluginKernelPermission("channels", []string{"add_source", "status", "list_sources"}, false) - setScript(`{"access_token":"MM-ACCESS-1","refresh_token":"MM-RF-1","nickname":"mm-user"}`) - st := poll(sid) - if st["state"] != "completed" { - t.Fatalf("行7 成功: 应 completed, 实得 %v", st) - } - if st["account"] != "mm-user" { - t.Fatalf("行7: account 应取 nickname, 实得 %v", st) - } - - // ---- 行 8: 其它错误 → failed 且带 error 文本 ---- - j = begin() - sid = j["session"].(map[string]interface{})["id"].(string) - setScript(`{"error":"server_error"}`) - st = poll(sid) - if st["state"] != "failed" { - t.Fatalf("行8 error: 应 failed, 实得 %v", st) - } - if msg, _ := st["message"].(string); !strings.Contains(msg, "server_error") { - t.Fatalf("行8: message 应带 error 文本, 实得 %v", st["message"]) + // 决策表全过 + if !h.runDecisionTable() { + t.Fatal("device-poll 决策表未全过(见 runDecisionTable 内的行号)") } } + +// mustChannelsSrc 读 bundled channels 源码。 +func mustChannelsSrc(t *testing.T) string { + t.Helper() + src, err := readBundledChannels(t) + if err != nil { + t.Fatal(err) + } + return src +} diff --git a/internal/core/channels_stress_test.go b/internal/core/channels_stress_test.go new file mode 100644 index 0000000..040a6d2 --- /dev/null +++ b/internal/core/channels_stress_test.go @@ -0,0 +1,251 @@ +package core + +import ( + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + + "llmsproxy/internal/config" + "llmsproxy/internal/lua" +) + +// mustBox 构造一个真 SecretBox(与运行期同一实现,密钥落 runtime dir)。 +func mustBox(t *testing.T, runtimeFile string) *config.SecretBox { + t.Helper() + box, err := config.NewSecretBox(runtimeFile) + if err != nil { + t.Fatal(err) + } + return box +} + +// channelsCore 起一个带 channels 插件、可指定源码与 box 的核心(压力测试用)。 +func channelsCore(t *testing.T, src string) (*Core, *lua.Plugins) { + t.Helper() + c := kernelTestCore(t) + ps := c.Plugins() + if err := ps.LoadSource("channels", src); err != nil { + t.Fatalf("加载 channels: %v", err) + } + ps.SetSecretBox(mustBox(t, filepath.Join(t.TempDir(), "runtime.json"))) + _ = ps.SetPluginKernelPermission("channels", []string{"add_source", "status", "list_sources"}, false) + return c, ps +} + +// TestChannelsConcurrentLogins 压力:并发走 device-poll 登录。 +// +// 插件是单 VM 单锁 —— 本测试验证的是**正确性下的并发**(会话互不串台、 +// 凭据互不可见、最终都 completed),真吞吐压力在 e2e 层(真二进制+HTTP 并发)。 +func TestChannelsConcurrentLogins(t *testing.T) { + var hits int32 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/oauth2/device/code": + n := atomic.AddInt32(&hits, 1) + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"device_code":"DEV-%d","user_code":"UC-%d","verification_uri":"https://example/activate","interval":1,"expires_in":600}`, n, n) + case "/oauth2/token": + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"access_token":"TOK","refresh_token":"RF","nickname":"u"}`) + default: + w.WriteHeader(404) + } + })) + defer ts.Close() + + _, ps := channelsCore(t, mustChannelsSrc(t)) + // 把 minimax 端点指到假上游 + if _, err := ps.Call("channels", "set_endpoints", map[string]interface{}{ + "platform": "minimax", + "device": map[string]interface{}{ + "code_url": ts.URL + "/oauth2/device/code", + "token_url": ts.URL + "/oauth2/token", + }, + }); err != nil { + t.Fatal(err) + } + + const rounds = 12 + var wg sync.WaitGroup + var failures int32 + for i := 0; i < rounds; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + out, err := ps.InvokeHandler("channels", "start_login", &lua.HandlerRequest{ + Method: "POST", Body: `{"platform":"minimax"}`, + }) + if err != nil { + t.Errorf("start_login #%d: %v", n, err) + atomic.AddInt32(&failures, 1) + return + } + j := out.JSON.(map[string]interface{}) + sess, _ := j["session"].(map[string]interface{}) + if sess == nil { + t.Errorf("start_login #%d 无会话: %v", n, j) + atomic.AddInt32(&failures, 1) + return + } + sid := sess["id"].(string) + for k := 0; k < 10; k++ { + out2, _ := ps.InvokeHandler("channels", "poll_login", &lua.HandlerRequest{ + Method: "GET", Query: map[string]string{"id": sid}, + }) + m2 := out2.JSON.(map[string]interface{}) + if m2["state"] == "completed" { + return + } + if m2["state"] == "failed" { + t.Errorf("会话 #%d failed: %v", n, m2) + atomic.AddInt32(&failures, 1) + return + } + } + t.Errorf("会话 #%d 10 次轮询未终态", n) + atomic.AddInt32(&failures, 1) + }(i) + } + wg.Wait() + if failures != 0 { + t.Fatalf("并发登录失败 %d 轮", failures) + } +} + +// TestChannelsClaimStress 压力:同一平台并发领签到(幂等 + 无登录态丢失)。 +func TestChannelsClaimStress(t *testing.T) { + var claims int32 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/checkin" { + atomic.AddInt32(&claims, 1) + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"ok":true,"credits":10}`) + return + } + w.WriteHeader(404) + fmt.Fprint(w, `{}`) + })) + defer ts.Close() + + _, ps := channelsCore(t, mustChannelsSrc(t)) + if _, err := ps.Call("channels", "set_endpoints", map[string]interface{}{ + "platform": "cookieish", + "checkin": map[string]interface{}{"url": ts.URL + "/v1/checkin"}, + }); err != nil { + t.Fatal(err) + } + if err := ps.SetSecret("channels", "channels:cookieish", "PASTE-TOKEN"); err != nil { + t.Fatal(err) + } + + const n = 30 + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(k int) { + defer wg.Done() + out, err := ps.InvokeHandler("channels", "claim_daily", &lua.HandlerRequest{ + Method: "POST", Body: `{"platform":"cookieish"}`, + }) + if err != nil { + t.Errorf("claim #%d: %v", k, err) + return + } + j := out.JSON.(map[string]interface{}) + if j["ok"] != true { + t.Errorf("claim #%d 未成功: %v", k, j) + } + }(i) + } + wg.Wait() + if got := atomic.LoadInt32(&claims); got != n { + t.Errorf("上游应收到 %d 次领取, 实得 %d", n, got) + } +} + +// TestLegacyVaultPlaintextLoad 旧通路:老版本(无加密)写的明文库在新版本 +// (带 box)下必须能读 —— 凭据升级不丢。落盘则会在下次写时整体重封为密文。 +func TestLegacyVaultPlaintextLoad(t *testing.T) { + dir := t.TempDir() + ps := lua.NewPlugins(nil, dir) + if err := ps.SetSecret("legacy", "channels:old", "PLAIN-OLD-TOKEN"); err != nil { + t.Fatal(err) + } + ps.SetSecretBox(mustBox(t, filepath.Join(dir, "runtime.json"))) + v, ok, err := ps.GetSecret("legacy", "channels:old") + if err != nil || !ok || v != "PLAIN-OLD-TOKEN" { + t.Fatalf("旧明文库读回失败: v=%q ok=%v err=%v", v, ok, err) + } + // 任意一次写触发整体重封 + if err := ps.SetSecret("legacy", "k2", "NEW"); err != nil { + t.Fatal(err) + } + raw, err := os.ReadFile(filepath.Join(dir, ".secrets", "legacy.json")) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(raw), "PLAIN-OLD-TOKEN") { + t.Fatalf("★ 重封后旧明文仍在盘上:\n%s", raw) + } + if !strings.Contains(string(raw), "enc:v1:") { + t.Fatalf("重封后应含密文:\n%s", raw) + } +} + +// TestLegacySourceSealedConfig 旧通路:config.yaml 里明文 api_key 的旧配置, +// 新版本加载后运行期拿到解密值、落盘被 seal(SealIfNeeded 语义)。 +func TestLegacySourceSealedConfig(t *testing.T) { + dir := t.TempDir() + cfgPath := filepath.Join(dir, "config.yaml") + yaml := "listen: 127.0.0.1:0\n" + + "gateway_keys:\n - sk-legacy-0001\n" + + "runtime_file: " + filepath.Join(dir, "runtime.json") + "\n" + + "adapter_dir: " + filepath.Join(dir, "adapters") + "\n" + + "sources:\n" + + " - name: legacy-src\n" + + " base_url: https://legacy.example.com/v1\n" + + " adapter: openai\n" + + " api_key: PLAIN-LEGACY-KEY\n" + + " max_concurrent: 2\n" + + " models:\n" + + " - id: m1\n" + + " kind: chat\n" + + " priority: 10\n" + if err := os.WriteFile(cfgPath, []byte(yaml), 0o644); err != nil { + t.Fatal(err) + } + cfg, err := config.Load(cfgPath) + if err != nil { + t.Fatal(err) + } + c, err := NewFromConfig(cfg) + if err != nil { + t.Fatal(err) + } + defer c.Close() + + got := c.Sources() + if len(got) != 1 || got[0].APIKey != "PLAIN-LEGACY-KEY" { + t.Fatalf("运行期 api_key 应解密: %#v", got) + } + raw, err := os.ReadFile(cfgPath) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(raw), "PLAIN-LEGACY-KEY") { + t.Fatalf("★ 旧明文 api_key 升级后仍明文落盘:\n%s", raw) + } + if !strings.Contains(string(raw), "enc:v1:") { + t.Fatalf("落盘应含 enc:v1 密文:\n%s", raw) + } + mk, err := os.ReadFile(filepath.Join(dir, "master.key")) + if err != nil || len(mk) != 64 { + t.Fatalf("master.key 应 64 hex, len=%d err=%v", len(mk), err) + } +} diff --git a/internal/core/core.go b/internal/core/core.go index 6fb9d5d..edb2570 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -131,6 +131,11 @@ func NewFromConfig(cfg *config.Config) (*Core, error) { // only appeared to work because seedKeys' Save() unsealed memory as a side // effect. Removing the redundant saves exposed the real ordering bug. cfg.AttachSecretBox(c.store.SecretBox()) + // Share that SAME box with the plugin credential vault: one master.key seals + // config.yaml, the runtime store, AND every .secrets/*.json. Set here (not at + // NewPlugins time) because the store — and thus the box — only exists now; + // plugins write secrets only during login flows, well after this point. + c.plugins.SetSecretBox(c.store.SecretBox()) hadPlaintextSecrets, err := cfg.UnsealSecrets(c.store.SecretBox()) if err != nil { return nil, fmt.Errorf("unseal secrets: %w", err) diff --git a/internal/core/device_harness_test.go b/internal/core/device_harness_test.go new file mode 100644 index 0000000..0bed988 --- /dev/null +++ b/internal/core/device_harness_test.go @@ -0,0 +1,180 @@ +package core + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "llmsproxy/internal/lua" +) + +// readBundledChannels 读内置 channels 插件源(供行为/变异测试共用)。 +func readBundledChannels(t *testing.T) (string, error) { + t.Helper() + return lua.ReadBundledPlugin("channels") +} + +// deviceHarness 是 device-poll 决策表的可复用测试台:假上游(脚本化逐次响应) +// + channels 插件加载指定源码 + minimax 端点指到假上游。 +// 行为测试(TestChannelsDevicePollBehavior)与变异测试共用它 —— 决策表只有 +// 一份,两处不会漂移。 +type deviceHarness struct { + t *testing.T + ps *lua.Plugins + ts *httptest.Server + step int32 + // script 持续被 HTTP handler 原子读取 + script atomic.Value // []scriptStep +} + +type scriptStep struct { + http int + body string +} + +func newDeviceHarness(t *testing.T, src string) *deviceHarness { + t.Helper() + h := &deviceHarness{t: t} + h.script.Store([]scriptStep{}) + + h.ts = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/oauth2/device/code": + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"device_code":"DEV1","user_code":"ABCD-EFGH","verification_uri":"https://example/activate","interval":1,"expires_in":600}`) + case "/oauth2/token": + i := int(atomic.AddInt32(&h.step, 1)) - 1 + script := h.script.Load().([]scriptStep) + w.Header().Set("Content-Type", "application/json") + if i >= len(script) { + w.WriteHeader(500) + fmt.Fprint(w, `{"error":"unexpected"}`) + return + } + if script[i].http != 0 && script[i].http != 200 { + w.WriteHeader(script[i].http) + } + fmt.Fprint(w, script[i].body) + default: + w.WriteHeader(404) + } + })) + t.Cleanup(h.ts.Close) + + c := kernelTestCore(t) + h.ps = c.Plugins() + if err := h.ps.LoadSource("channels", src); err != nil { + t.Fatalf("加载: %v", err) + } + if _, err := h.ps.Call("channels", "set_endpoints", map[string]interface{}{ + "platform": "minimax", + "device": map[string]interface{}{ + "code_url": h.ts.URL + "/oauth2/device/code", + "token_url": h.ts.URL + "/oauth2/token", + }, + }); err != nil { + t.Fatalf("set_endpoints: %v", err) + } + return h +} + +// setScript 设置 token 端点的逐次响应脚本。`400 {...}` 形式表示非 200。 +func (h *deviceHarness) setScript(steps ...string) { + var out []scriptStep + for _, b := range steps { + code := 200 + switch { + case strings.HasPrefix(b, "400 "): + code, b = 400, b[4:] + case strings.HasPrefix(b, "403 "): + code, b = 403, b[4:] + } + out = append(out, scriptStep{http: code, body: b}) + } + h.script.Store(out) + atomic.StoreInt32(&h.step, 0) +} + +func (h *deviceHarness) begin() map[string]interface{} { + out, err := h.ps.InvokeHandler("channels", "start_login", &lua.HandlerRequest{ + Method: "POST", Body: `{"platform":"minimax"}`, + }) + if err != nil { + h.t.Fatal(err) + } + return out.JSON.(map[string]interface{}) +} + +func (h *deviceHarness) poll(id string) map[string]interface{} { + out, _ := h.ps.InvokeHandler("channels", "poll_login", &lua.HandlerRequest{ + Method: "GET", Query: map[string]string{"id": id}, + }) + return out.JSON.(map[string]interface{}) +} + +// runDecisionTable 跑完整 8 行决策表,全部符合预期返回 true。 +// 这是行为契约的单一事实来源:行为测试用它断言通过;变异测试用它杀变异体。 +func (h *deviceHarness) runDecisionTable() bool { + ok := func(cond bool) bool { return cond } + st := func(steps ...string) map[string]interface{} { + h.setScript(steps...) + j := h.begin() + sid := j["session"].(map[string]interface{})["id"].(string) + return h.poll(sid) + } + + // 行1: 200+pending → 等 + if j := st(`{"status":"pending"}`); j["state"] != "pending" { + return false + } + // 行2: 200+slow_down → 等 + interval+5 + h.setScript(`{"status":"slow_down"}`) + j := h.begin() + sid := j["session"].(map[string]interface{})["id"].(string) + if j2 := h.poll(sid); j2["state"] != "pending" { + return false + } else if iv, _ := j2["interval"].(float64); iv < 6 { + return false + } + // 行3: 200+denied → failed,message 区分于通用错误(杀"denied 不识别"变异体) + if j := st(`{"status":"denied"}`); j["state"] != "failed" { + return false + } else if msg, _ := j["message"].(string); !strings.Contains(msg, "用户拒绝") { + return false + } + // 行4: 200+expired_token → failed,message 带过期字样 + if j := st(`{"status":"expired_token"}`); j["state"] != "failed" { + return false + } else if msg, _ := j["message"].(string); !strings.Contains(msg, "过期") { + return false + } + // 行5: 400+authorization_pending → 等 + if j := st(`400 {"error":"authorization_pending"}`); j["state"] != "pending" { + return false + } + // 行6: 403+slow_down → 等 + interval+5 + h.setScript(`403 {"error":"slow_down"}`) + j = h.begin() + sid = j["session"].(map[string]interface{})["id"].(string) + if j2 := h.poll(sid); j2["state"] != "pending" { + return false + } else if iv, _ := j2["interval"].(float64); iv < 6 { + return false + } + // 行7: 200+access_token → completed(需先授权 add_source) + _ = h.ps.SetPluginKernelPermission("channels", []string{"add_source", "status", "list_sources"}, false) + h.setScript(`{"access_token":"MM-ACCESS-1","refresh_token":"MM-RF-1","nickname":"mm-user"}`) + j = h.begin() + sid = j["session"].(map[string]interface{})["id"].(string) + if j2 := h.poll(sid); j2["state"] != "completed" || j2["account"] != "mm-user" { + return false + } + // 行8: 其它 error → failed 带文本 + if j := st(`{"error":"server_error"}`); j["state"] != "failed" { + return false + } + return ok(true) +} diff --git a/internal/core/device_mutate_test.go b/internal/core/device_mutate_test.go new file mode 100644 index 0000000..a6d088c --- /dev/null +++ b/internal/core/device_mutate_test.go @@ -0,0 +1,69 @@ +package core + +import ( + "strings" + "testing" +) + +// TestDevicePollMutants 是 device-poll 执行器的变异测试。 +// +// 用户要求:移植要照参考实现的行为形成测试项,且要变异验证测试真的能杀掉 +// 「行为走样」的代码。每个用例把 channels 源里的一个决策分支做语义变异, +// 断言决策表**会失败**(变异体被杀)。变异体幸存 = 该分支没有覆盖。 +func TestDevicePollMutants(t *testing.T) { + src, err := readBundledChannels(t) + if err != nil { + t.Fatal(err) + } + + muts := []struct { + name string + from string + to string + }{ + { + name: "slow_down 不再加 interval", + from: `sess.device_interval = (sess.device_interval or d.interval or 5) + 5`, + to: `sess.device_interval = (sess.device_interval or d.interval or 5)`, + }, + { + name: "denied 被当成继续等待", + from: `if resp.status == 200 and (status == "denied" or status == "access_denied") then`, + to: `if resp.status == 200 and (status == "denied_x" or status == "access_denied") then`, + }, + { + name: "expired_token 不被识别", + from: `if resp.status == 200 and (status == "expired" or status == "expired_token") then`, + to: `if resp.status == 200 and (status == "expired" or status == "expired_token_x") then`, + }, + { + name: "authorization_pending 被当成失败", + from: `if ecode == "authorization_pending" then return false end`, + to: `if ecode == "authorization_pending_x" then return false end`, + }, + { + name: "成功分支被禁用", + from: `if resp.status == 200 and rec.access_token then`, + to: `if resp.status == 200 and rec.access_token and false then`, + }, + } + + for _, mu := range muts { + t.Run(mu.name, func(t *testing.T) { + if !strings.Contains(src, mu.from) { + t.Fatalf("锚点丢失,先更新实现与锚点:%q", mu.from) + } + mutated := strings.Replace(src, mu.from, mu.to, 1) + if mutated == src { + t.Fatal("变异未生效") + } + // 基准必须全过(测试自身有效),变异体必须被杀(有覆盖) + if !newDeviceHarness(t, src).runDecisionTable() { + t.Fatal("基准决策表未全过 —— 测试自身失效") + } + if newDeviceHarness(t, mutated).runDecisionTable() { + t.Fatalf("★ 变异体幸存(%s)—— 该分支无覆盖", mu.name) + } + }) + } +} diff --git a/internal/lua/plugin_secrets.go b/internal/lua/plugin_secrets.go index 2928003..aa77237 100644 --- a/internal/lua/plugin_secrets.go +++ b/internal/lua/plugin_secrets.go @@ -43,6 +43,24 @@ import ( // secretsDirName is the vault directory inside plugin_dir. The leading dot keeps // it out of a casual `ls` of the plugin directory, which is where operators look // for "what plugins do I have" — the vault is not a plugin. +// secretSealer 是保险库落盘加密所需的最小能力面。 +// +// 不直接 import config.SecretBox 的原因与 KernelAPI 相同:保持依赖注入的缝, +// 测试可以用桩;core 侧把 *config.SecretBox 注进来(Encrypt/Decrypt 语义与 +// config.yaml 的凭据完全一致 —— 同一把 master key、同一种 enc:v1 前缀)。 +type secretSealer interface { + Encrypt(plain string) (string, error) + Decrypt(v string) (string, error) +} + +// SetSecretBox 注入落盘加密器。必须在任何插件写凭据之前调用(core 启动序里 +// 在 NewPlugins 之后立即做)。传 nil 合法(=不加密,测试环境)。 +func (ps *Plugins) SetSecretBox(box secretSealer) { + ps.mu.Lock() + ps.secretBox = box + ps.mu.Unlock() +} + const secretsDirName = ".secrets" // secretsDir returns the vault directory (empty when there is no plugin dir, in @@ -105,6 +123,18 @@ func (ps *Plugins) loadSecretsLocked(plugin string) (map[string]string, error) { if err := json.Unmarshal(b, &m); err != nil { return nil, fmt.Errorf("vault for plugin %s is unreadable: %w", plugin, err) } + // Decrypt each value. Decrypt passes values without the enc:v1 prefix through + // unchanged, so vaults written by an older plaintext build keep loading — + // and get re-sealed on the next write (SetSecret rewrites the whole file). + if ps.secretBox != nil { + for k, v := range m { + plain, err := ps.secretBox.Decrypt(v) + if err != nil { + return nil, fmt.Errorf("vault for plugin %s: key %q: %w (master key changed?)", plugin, k, err) + } + m[k] = plain + } + } return m, nil } @@ -126,7 +156,23 @@ func (ps *Plugins) writeSecretsLocked(plugin string, m map[string]string) error // vault created before this mode was enforced would keep its old permissions. _ = os.Chmod(dir, 0o700) - b, err := json.MarshalIndent(m, "", " ") + // Encrypt-at-rest: each VALUE is sealed with the deployment's master key + // (same enc:v1 form as config.yaml credentials). Keys stay plaintext so the + // file stays diffable/debuggable without exposing token bytes. A nil box + // writes plaintext — tests only; core always injects a real box. + sealed := make(map[string]string, len(m)) + for k, v := range m { + if ps.secretBox != nil { + enc, err := ps.secretBox.Encrypt(v) + if err != nil { + return fmt.Errorf("seal secret %q: %w", k, err) + } + sealed[k] = enc + } else { + sealed[k] = v + } + } + b, err := json.MarshalIndent(sealed, "", " ") if err != nil { return err } diff --git a/internal/lua/plugins.go b/internal/lua/plugins.go index 88f4b0a..6adb017 100644 --- a/internal/lua/plugins.go +++ b/internal/lua/plugins.go @@ -308,6 +308,12 @@ type Plugins struct { // through this interface, never through a direct import (core depends on // lua, not the other way round). kernelAPI KernelAPI + // secretBox seals the credential vault at rest. nil = write plaintext (tests + // / embedders with no runtime dir). Injected by core via SetSecretBox: it is + // the deployment's ONE master key, the same one that seals config.yaml, so a + // lost master.key loses the plugin tokens too — loud, consistent, not a + // second key to manage. + secretBox secretSealer // kernelPerms records what the OPERATOR granted each plugin. Separate from // what a plugin DECLARES it wants (Plugin.KernelDeclared): a declaration is // a request, a grant is a privilege. Persisted in .access.json. diff --git a/internal/lua/secrets_crypto_test.go b/internal/lua/secrets_crypto_test.go new file mode 100644 index 0000000..28fe838 --- /dev/null +++ b/internal/lua/secrets_crypto_test.go @@ -0,0 +1,227 @@ +package lua + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "os" + "path/filepath" + "strings" + "sync" + "testing" +) + +// ---- 测试用加密桩:与 config.SecretBox 同语义(AEAD + enc:v1 前缀)---- + +const testEncPrefix = "enc:v1:" + +// realBox:真 AEAD-GCM(与 config.SecretBox 同原语),行为一致。 +type realBox struct { + aead cipher.AEAD +} + +func newRealBox(t *testing.T) *realBox { + t.Helper() + digest := sha256.Sum256([]byte("test-master-key")) + block, err := aes.NewCipher(digest[:]) + if err != nil { + t.Fatal(err) + } + aead, err := cipher.NewGCM(block) + if err != nil { + t.Fatal(err) + } + return &realBox{aead: aead} +} + +func (b *realBox) Encrypt(plain string) (string, error) { + nonce := make([]byte, b.aead.NonceSize()) + if _, err := rand.Read(nonce); err != nil { + return "", err + } + sealed := b.aead.Seal(nonce, nonce, []byte(plain), nil) + return testEncPrefix + b64RawURL(sealed), nil +} + +func (b *realBox) Decrypt(v string) (string, error) { + if !strings.HasPrefix(v, testEncPrefix) { + return v, nil + } + raw, err := b64RawURLDecode(strings.TrimPrefix(v, testEncPrefix)) + if err != nil { + return v, err + } + n := b.aead.NonceSize() + if len(raw) < n { + return v, os.ErrInvalid + } + plain, err := b.aead.Open(nil, raw[:n], raw[n:], nil) + if err != nil { + return v, err + } + return string(plain), nil +} + +// failBox 永远解密失败(模拟 master.key 被换/丢失)。 +type failBox struct{} + +func (failBox) Encrypt(plain string) (string, error) { return testEncPrefix + "AAAA", nil } +func (failBox) Decrypt(v string) (string, error) { + if !strings.HasPrefix(v, testEncPrefix) { + return v, nil + } + return v, os.ErrPermission +} + +func b64RawURL(b []byte) string { return base64.RawURLEncoding.EncodeToString(b) } +func b64RawURLDecode(s string) ([]byte, error) { + return base64.RawURLEncoding.DecodeString(s) +} + +// TestSecretVaultEncryptedAtRest 验证保险库落盘是密文(enc:v1 前缀), +// 且读回时解密为原值。明文 token 绝不能出现在磁盘字节里。 +func TestSecretVaultEncryptedAtRest(t *testing.T) { + dir := t.TempDir() + ps := NewPlugins(nil, dir) + ps.SetSecretBox(newRealBox(t)) + + if err := ps.SetSecret("channels", "channels:minimax", "SUPER-TOKEN-123"); err != nil { + t.Fatalf("SetSecret: %v", err) + } + + // 磁盘字节必须不含明文 + raw, err := os.ReadFile(filepath.Join(dir, ".secrets", "channels.json")) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(raw), "SUPER-TOKEN-123") { + t.Fatalf("★ 保险库明文落盘! 文件内容含 token:\n%s", raw) + } + if !strings.Contains(string(raw), testEncPrefix) { + t.Fatalf("落盘值应带 %s 前缀(统一密文形态):\n%s", testEncPrefix, raw) + } + + // 读回是原值 + v, ok, err := ps.GetSecret("channels", "channels:minimax") + if err != nil || !ok || v != "SUPER-TOKEN-123" { + t.Fatalf("round-trip 失败: v=%q ok=%v err=%v", v, ok, err) + } + + // 文件权限 0600、目录 0700 + if fi, err := os.Stat(filepath.Join(dir, ".secrets", "channels.json")); err != nil || fi.Mode().Perm() != 0o600 { + t.Fatalf("vault 文件权限应 0600, 实得 %v (err=%v)", fi.Mode(), err) + } + if fi, err := os.Stat(filepath.Join(dir, ".secrets")); err != nil || fi.Mode().Perm() != 0o700 { + t.Fatalf("vault 目录权限应 0700, 实得 %v (err=%v)", fi.Mode(), err) + } +} + +// TestSecretVaultPlaintextFallbackAndUpgrade:旧明文库可读(向后兼容), +// 且下一次写会整体升级为密文。 +func TestSecretVaultPlaintextFallbackAndUpgrade(t *testing.T) { + dir := t.TempDir() + ps := NewPlugins(nil, dir) + // 1) 无 box 时写入 = 旧版本形态(明文) + if err := ps.SetSecret("channels", "k", "OLD-PLAIN"); err != nil { + t.Fatal(err) + } + // 2) 换上有 box 的新版本:旧明文必须能读出来 + ps.SetSecretBox(newRealBox(t)) + v, ok, err := ps.GetSecret("channels", "k") + if err != nil || !ok || v != "OLD-PLAIN" { + t.Fatalf("旧明文读回应原值: v=%q ok=%v err=%v", v, ok, err) + } + // 3) 任意一次写(哪怕写别的键)会把整个文件升级为密文 + if err := ps.SetSecret("channels", "k2", "NEW"); err != nil { + t.Fatal(err) + } + raw, _ := os.ReadFile(filepath.Join(dir, ".secrets", "channels.json")) + if strings.Contains(string(raw), "OLD-PLAIN") { + t.Fatalf("★ 升级写后旧明文值仍在盘上:\n%s", raw) + } + var m map[string]string + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(m["k"], testEncPrefix) { + t.Fatalf("旧值应被重封为密文: %v", m["k"]) + } + // 4) 两个值都还能读回 + for key, want := range map[string]string{"k": "OLD-PLAIN", "k2": "NEW"} { + if v, _, _ := ps.GetSecret("channels", key); v != want { + t.Fatalf("升级后 %q: got %q want %q", key, v, want) + } + } +} + +// TestSecretVaultWrongMasterKey:master.key 换掉后读库必须**报错**而不是静默 +// 返回空 —— 空会让插件重新登录并覆盖文件,把可恢复问题变成凭据丢失。 +func TestSecretVaultWrongMasterKey(t *testing.T) { + dir := t.TempDir() + ps := NewPlugins(nil, dir) + ps.SetSecretBox(newRealBox(t)) + if err := ps.SetSecret("channels", "k", "SECRET"); err != nil { + t.Fatal(err) + } + // 换一把钥匙(模拟 master.key 丢失重建) + ps.SetSecretBox(failBox{}) + if _, _, err := ps.GetSecret("channels", "k"); err == nil { + t.Fatal("★ 换 master key 后读库未报错 —— 会静默吞凭据") + } +} + +// ---- 并发压力:多插件同时读写各自的库 ---- + +// TestSecretVaultConcurrentPlugins 压力:8 个插件 × 每插件 2 goroutine 并发 +// set/get/delete 自己的键。断言:无数据串台(跨插件不可见)、无丢写、文件数正确。 +func TestSecretVaultConcurrentPlugins(t *testing.T) { + dir := t.TempDir() + ps := NewPlugins(nil, dir) + ps.SetSecretBox(newRealBox(t)) + + const plugins = 8 + const perG = 50 + var wg sync.WaitGroup + for p := 0; p < plugins; p++ { + name := "plug" + string(rune('A'+p)) + for g := 0; g < 2; g++ { + wg.Add(1) + go func(name string, g int) { + defer wg.Done() + for i := 0; i < perG; i++ { + key := name + ":k" + string(rune('a'+g)) + if err := ps.SetSecret(name, key, name+"-token"); err != nil { + t.Errorf("SetSecret(%s,%s): %v", name, key, err) + return + } + v, ok, err := ps.GetSecret(name, key) + if err != nil || !ok || v != name+"-token" { + t.Errorf("GetSecret(%s,%s)=%q ok=%v err=%v", name, key, v, ok, err) + return + } + // 跨插件必须不可见 + other := "plug" + string(rune('A'+(p+1)%plugins)) + if _, ok, _ := ps.GetSecret(other, key); ok { + t.Errorf("★ 凭据串台: %s 的键在 %s 可见", name, other) + return + } + if g == 1 && i%10 == 9 { + _ = ps.DeleteSecret(name, key) + } + } + }(name, g) + } + } + wg.Wait() + + entries, err := os.ReadDir(filepath.Join(dir, ".secrets")) + if err != nil { + t.Fatal(err) + } + if len(entries) > plugins { + t.Errorf("vault 文件数 %d 超过插件数 %d(应一插件一文件)", len(entries), plugins) + } +}