diff --git a/internal/core/channels_device_test.go b/internal/core/channels_device_test.go new file mode 100644 index 0000000..204c6cb --- /dev/null +++ b/internal/core/channels_device_test.go @@ -0,0 +1,181 @@ +package core + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "llmsproxy/internal/lua" +) + +// TestChannelsDevicePollBehavior 是 device-poll 流的行为对齐测试。 +// +// 决策表逐行搬自参考实现 pollMinimaxDeviceToken(minimax-oauth.ts), +// 上游用假服务器模拟,逐次响应由测试控制。断言的是**行为**(会话状态推进), +// 不是实现细节。 +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() + 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) + } + + // ---- 行 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"]) + } +} diff --git a/internal/lua/plugins/channels.lua b/internal/lua/plugins/channels.lua index 3af4c3d..b50a498 100644 --- a/internal/lua/plugins/channels.lua +++ b/internal/lua/plugins/channels.lua @@ -50,6 +50,10 @@ local plugin = { -- 申请的内核权限。设置页会把"申请中但未授权"标红。 plugin.kernel = { "add_source", "list_sources", "probe_all", "reset_health", "status" } +-- WebUI 可调用的导出函数(POST /api/plugins/channels/call/{fn})。 +-- set_endpoints 让运维把渠道端点指向自建代理/镜像;测试用它打假上游。 +plugin.exports = { "set_endpoints" } + -- ============================================================================ @@ -208,6 +212,96 @@ local function bodyOf(req) return v end +-- ============================================================================ +-- device-poll 通用执行器(移植自参考实现 minimax-oauth.ts / cline-oauth.ts) +-- ============================================================================ +-- 行为逐字对齐参考实现,测试用例也照它搬(见 channels_device_test.go): +-- +-- 1. PKCE S256:verifier = 随机字节 base64url;challenge = SHA256(verifier) +-- 的 base64url(参考 createMinimaxPkce)。 +-- 2. 轮询决策分支的**顺序**就是行为(参考 pollMinimaxDeviceToken): +-- HTTP 200 + status=pending → 继续等 +-- HTTP 200 + status=slow_down → interval += 5s,继续等 +-- HTTP 200 + status=denied / +-- access_denied → 会话 failed(用户拒绝) +-- HTTP 200 + status=expired / +-- expired_token → 会话 failed(码过期) +-- 非 200 + error=authorization_pending → 继续等(OAuth 标准形态) +-- 非 200 + error=slow_down → interval += 5s,继续等 +-- HTTP 200(无上述状态) → 成功,提取 access_token +-- 其它 → 会话 failed(带 error 文本) +-- ⚠️ 「两种还在等」都要认:形一是 200+status 字段,形二是非 200+error 字段。 +-- 3. 轮询由页面驱动(poll_login 每次调用推进一步),不在 Lua 里阻塞循环: +-- 插件单 VM 单锁,阻塞循环会卡住本插件的全部钩子。 + +-- b64url:内核的 base64_encode 是标准 base64;PKCE 要 base64url(无 +/= )。 +local function b64url(s) + return (base64_encode(s):gsub("+", "-"):gsub("/", "_"):gsub("=+$", "")) +end + +-- hexToBin:把 hex 字符串还原成二进制字符串(PKCE challenge 输入用)。 +local function hexToBin(h) + return (h:gsub("..", function(cc) return string.char(tonumber(cc, 16)) end)) +end + +-- devicePkce 生成一次授权的 verifier/challenge 对(SHA256 S256)。 +-- verifier 自造:熵拼接后哈希。不依赖文件后段才定义的 newToken —— Lua 词法 +-- 作用域下那会是 nil 全局调用。熵用 os.time + math.random 两次哈希拼接。 +local function devicePkce() + local v = sha256_hex(tostring(os.time and os.time() or 0) .. math.random(1e12)) + .. sha256_hex(tostring(math.random(1e12))) + return v, b64url(hexToBin(sha256_hex(v))) +end + +-- deviceOneStep 推进一次设备码轮询,返回 (done, payload)。 +-- done=true payload={ ok=true, access_token=.., refresh_token=.., account=.. } +-- 或 { ok=false, reason="denied"|"expired", message=.. } +-- done=false payload=nil(还在等:pending / slow_down) +-- sess 携带 device 上下文(device_code/verifier/interval),被本函数原地更新 +--(slow_down 后 interval 增大,与参考一致)。 +local function deviceOneStep(plat, sess) + local d = plat.device + local body = (d.token_body + :gsub("{device_code}", sess.device_code) + :gsub("{verifier}", sess.device_verifier or "") + :gsub("{challenge}", sess.device_challenge or "")) + local resp, err = __llmsproxy_http({ + url = d.token_url, + method = "POST", + headers = { ["Accept"] = "application/json", ["Content-Type"] = "application/x-www-form-urlencoded" }, + body = body, + }) + if err then return false end -- 网络抖动不算失败:与参考一致(pending 语义) + local ok, rec = pcall(json.decode, resp.body or "") + if not ok or type(rec) ~= "table" then rec = {} end + local status = rec.status + local ecode = rec.error + + -- 形态一:HTTP 200 + status + if resp.status == 200 and status == "pending" then return false end + if resp.status == 200 and status == "slow_down" then + sess.device_interval = (sess.device_interval or d.interval or 5) + 5 + return false + end + if resp.status == 200 and (status == "denied" or status == "access_denied") then + return true, { ok = false, reason = "denied", message = "用户拒绝了授权" } + end + if resp.status == 200 and (status == "expired" or status == "expired_token") then + return true, { ok = false, reason = "expired", message = "设备码已过期,请重新发起登录" } + end + -- 形态二:非 200 + error(OAuth 标准形态) + if ecode == "authorization_pending" then return false end + if ecode == "slow_down" then + sess.device_interval = (sess.device_interval or d.interval or 5) + 5 + return false + end + if resp.status == 200 and rec.access_token then + return true, { ok = true, access_token = rec.access_token, + refresh_token = rec.refresh_token, account = rec.nickname or rec.sub or "" } + end + return true, { ok = false, reason = "error", message = "授权失败: " .. tostring(ecode or ("HTTP " .. tostring(resp.status))) } +end + -- platformById 查平台定义;未知 id 返回 nil(调用方必须处理,而不是拿到空表) local function platformById(id) for _, p in ipairs(PLATFORMS) do @@ -386,6 +480,38 @@ function plugin.start_login(req) out.view_url = bs.view_url logEvent("服务端浏览器已就绪 " .. plat.id) end + elseif plat.flow == "device-poll" then + -- 设备码:申请码,把 user_code + 授权 URL 给用户,verifier 存进会话。 + -- 行为对齐参考 startMinimaxDeviceAuthorization:PKCE S256。 + local verifier, challenge = devicePkce() + local body = (plat.device.code_body:gsub("{challenge}", challenge)) + local resp, err = __llmsproxy_http({ + url = plat.device.code_url, + method = "POST", + headers = { ["Accept"] = "application/json", ["Content-Type"] = "application/x-www-form-urlencoded" }, + body = body, + }) + if err then + sess.state = "failed" + out.error = "设备码申请失败: " .. err + return { status = 200, json = out } + end + local ok, rec = pcall(json.decode, resp.body or "") + if not ok or type(rec) ~= "table" or not rec.device_code then + sess.state = "failed" + out.error = "设备码响应无法解析(HTTP " .. tostring(resp.status) .. ")" + return { status = 200, json = out } + end + sess.device_code = rec.device_code + sess.device_verifier = verifier + sess.device_challenge = challenge + sess.device_interval = tonumber(rec.interval) or plat.device.interval or 5 + sess.user_code = rec.user_code + -- 授权页 URL:优先用 verification_uri_complete(免手输 user_code) + out.url = rec.verification_uri_complete or rec.verification_uri or rec.verification_url or plat.login_url + out.session.user_code = rec.user_code + out.session.interval = sess.device_interval + logEvent("设备码已申请 " .. plat.id .. " user_code=" .. tostring(rec.user_code)) else -- ①②:URL 交给用户自己的浏览器;callback 流程的 redirect_uri 指向本插件挂载点 if plat.flow == "callback" then @@ -398,12 +524,37 @@ function plugin.start_login(req) end -- plugin.poll_login(session_id) 由页面轮询:会话是否完成。 +-- +-- 对 device-poll 会话,每次轮询**推进一步**设备码换 token(deviceOneStep): +-- 轮询由页面驱动而不在 Lua 里阻塞,因为插件是单 VM 单锁,阻塞循环会卡死 +-- 本插件全部钩子。成功则落凭据+接入源(走 complete_login)。 function plugin.poll_login(req) local id = (req and req.query and req.query.id) or "" sweepSessions() local s = sessions()[id] if not s then return { status = 200, json = { state = "gone" } } end - return { status = 200, json = { state = s.state, account = s.account, message = s.message } } + + if s.flow == "device-poll" and s.state == "pending" and s.device_code then + local plat = platformById(s.platform) + local done, payload = deviceOneStep(plat, s) + if done and payload then + if payload.ok then + -- 成功:落凭据 + 接入源 + local res = plugin.complete_login({ platform = s.platform, credential = payload.access_token, account = payload.account }) + if res.json and res.json.ok then + s.state = "completed" + s.account = payload.account + else + s.state = "failed" + s.message = (res.json and res.json.error) or "接入源失败" + end + else + s.state = "failed" + s.message = payload.message + end + end + end + return { status = 200, json = { state = s.state, account = s.account, message = s.message, interval = s.device_interval } } end -- plugin.complete_login(args) 由页面/回调调用:登录成功,存凭据并接入源。 @@ -554,6 +705,37 @@ plugin.http = { public = false, roles = { "admin" }, description = "OAuth 回调" }, } +-- set_endpoints:改写某平台的端点声明(device.code_url/token_url、checkin.url、 +-- base_url)。两个用途: +-- 1. 运维把某渠道指向自建代理/镜像(厂商被墙、企业出口、审计要求) +-- 2. 测试把端点指到假上游,行为测试才能不打真厂商 +-- 只允许改已知字段,未知键拒绝 —— 端点声明是协议形状的一部分,静默加字段 +-- 会破坏后续逻辑对形状的假设。 +function plugin.set_endpoints(req) + local args = bodyOf(req) + if args.platform == nil then args = req end + local plat = platformById(args and args.platform) + if not plat then return { status = 400, json = { error = "未知平台" } } end + local allowed = { base_url = true, login_url = true } + local changed = {} + for k in pairs(allowed) do + if args[k] then plat[k] = tostring(args[k]) changed[#changed+1] = k end + end + if type(args.device) == "table" and plat.device then + for _, k in ipairs({ "code_url", "token_url" }) do + if args.device[k] then plat.device[k] = tostring(args.device[k]) changed[#changed+1] = "device." .. k end + end + end + if type(args.checkin) == "table" and plat.checkin then + for _, k in ipairs({ "url" }) do + if args.checkin[k] then plat.checkin[k] = tostring(args.checkin[k]) changed[#changed+1] = "checkin." .. k end + end + end + if #changed == 0 then return { status = 400, json = { error = "无可识别的端点字段" } } end + logEvent("端点已改写 " .. plat.id .. ": " .. table.concat(changed, ",")) + return { status = 200, json = { ok = true, changed = changed } } +end + function plugin.list_platforms(req) local out = {} for _, p in ipairs(PLATFORMS) do