feat(weixin-ilink,gateway): 会话过期显式化 — 不再静默变哑/假 OK 重连,面板可见「需重扫」
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- 适配器新增过期状态机(plugins/channels/weixin_ilink/session_state.go):getupdates/sendmessage/getuploadurl 收到 errcode=-14 即 markExpired(ERROR 仅一次,只带 bot_id,token 永不入日志),扫码登录拿到新 token 或配置换新 token 时清除(惰性比对)
- Connect/HealthCheck 在过期时返回 ErrSessionExpired(可与 ErrNotConnected 区分):不再对死 token 设 connected=true 报假成功;发送路径同样返回该错误而不是笼统 ErrNotConnected
- 网关:POST /admin/channels/open 遇过期 → 409 session_expired + 明确文案(不再吞错回 200);GET /admin/channels 摘要新增 needsRescan/lastError(经可选接口类型断言读取,取不到不伪造)
- channelbridge:读循环非取消退出时把自身记账的 connected 置 false(热重载被替换者退出不抹掉新实例)
- login/login_admin:过期时 StartLogin 允许重新出码、LoginStatus 如实报 failed(原先可能把死 token 当 confirmed 重新落盘)
- Docs/api-client v0.39;新增 5 组测试 + 变异验证(去掉标记/恢复假成功/网关不填字段 均 FAIL)
This commit is contained in:
AskaEth committed 2026-10-03 18:00:01 +08:00
1 parent 2092094d0a
commit cdc370dc82
14 files changed
+975 -30

No files matched your search

@@ -22,11 +22,41 @@ import (
// deliver through the same pipeline as push-style adapters. Message processing
// uses the bridge-lifetime context (not readCtx) so a reload leaves in-flight
// messages alone.
//
// When the read path returns with readCtx still alive, the adapter stopped
// reading on its own — today that is the weixin-iLink expired session
// (errcode=-14, see plugins/channels/weixin_ilink/session_state.go), which used to
// leave the bridge claiming the channel was connected while it could no longer
// deliver anything. A cancelled readCtx is a deliberate stop (bridge Stop or a
// hot reload of this name) and is handled by the caller that cancelled it.
func (b *Bridge) readLoop(a adapter.PlatformAdapter, readCtx context.Context) {
defer b.wg.Done()
a.ReadMessages(readCtx, func(msg adapter.UnifiedMessage) {
b.onMessage(b.ctx, msg, a)
})
if readCtx.Err() == nil {
b.markReadStopped(a)
}
}
// markReadStopped records that an adapter's read loop ended by itself while the
// adapter reports no live session, so the per-name accounting (IsConnected /
// ConnectedCount) stops claiming a channel that cannot deliver. It only ever
// clears the flag — never sets it — so it cannot make a connected channel look
// alive, and it is a no-op when the adapter still reports a live session (a
// reconnect that already happened is never undone by an older loop's exit).
func (b *Bridge) markReadStopped(a adapter.PlatformAdapter) {
if a.IsConnected() {
return
}
name := a.Name()
b.mu.Lock()
defer b.mu.Unlock()
if !b.connected[name] {
return
}
b.connected[name] = false
b.log.Warn("channel read loop ended, adapter is no longer connected", "channel", name)
}
// onMessage is the adapter-facing inbound callback. It runs the pipeline in a
@@ -0,0 +1,128 @@
package channelbridge
import (
"context"
"sync"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)
// deadSessionAdapter mimics a channel whose session dies while the bridge is
// running — today the weixin-iLink expired bot token (iLink errcode=-14): Connect
// reports success, the read loop then returns on its own (nothing cancelled it)
// and IsConnected reports false from that point on.
type deadSessionAdapter struct {
name string
mu sync.Mutex
up bool
}
func (a *deadSessionAdapter) Name() string { return a.name }
func (a *deadSessionAdapter) Config() map[string]any { return map[string]any{"bot_token": "tok"} }
func (a *deadSessionAdapter) Connect(context.Context) error {
a.mu.Lock()
a.up = true
a.mu.Unlock()
return nil
}
func (a *deadSessionAdapter) Disconnect(context.Context) error {
a.mu.Lock()
a.up = false
a.mu.Unlock()
return nil
}
func (a *deadSessionAdapter) IsConnected() bool {
a.mu.Lock()
defer a.mu.Unlock()
return a.up
}
func (a *deadSessionAdapter) HealthCheck(context.Context) error { return adapter.ErrNotConnected }
func (a *deadSessionAdapter) ReadMessages(context.Context, func(adapter.UnifiedMessage)) {
// The session died: the read loop ends by itself, no cancellation involved.
a.mu.Lock()
a.up = false
a.mu.Unlock()
}
func (a *deadSessionAdapter) SendMessage(context.Context, adapter.UnifiedResponse) error {
return adapter.ErrNotConnected
}
func (a *deadSessionAdapter) SendProactive(context.Context, adapter.UnifiedMessage) error {
return adapter.ErrNotConnected
}
func (a *deadSessionAdapter) SetMessageHandler(adapter.MessageHandler) {}
func (a *deadSessionAdapter) Capabilities() adapter.Capabilities {
return adapter.DefaultCapabilities()
}
func (a *deadSessionAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) {
return nil, nil
}
// signaledAdapter is a fakeAdapter whose read loop reports when it has returned,
// so a test can wait for a replaced adapter's exit deterministically.
type signaledAdapter struct {
*fakeAdapter
exited chan struct{}
}
func (a *signaledAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) {
<-ctx.Done()
close(a.exited)
}
// TestReadLoopExitClearsConnectedFlag is the regression for "the adapter goes mute
// while the bridge still claims it is connected": the read loop ends with the
// session down and nothing cancelled it, so IsConnected/ConnectedCount must stop
// reporting a live channel.
func TestReadLoopExitClearsConnectedFlag(t *testing.T) {
t.Parallel()
b := newReloadBridge(t, nil)
ad := &deadSessionAdapter{name: "weixin-ilink"}
b.Add(ad)
b.Start()
waitDeadline := time.Now().Add(2 * time.Second)
for time.Now().Before(waitDeadline) {
if !b.IsConnected("weixin-ilink") {
if n := b.ConnectedCount(); n != 0 {
t.Fatalf("ConnectedCount() = %d, want 0 after the only read loop died", n)
}
return
}
time.Sleep(time.Millisecond)
}
t.Fatal("bridge still reports weixin-ilink as connected after its read loop ended")
}
// TestReplacedAdapterExitKeepsReplacementConnected guards the other direction: the
// read-loop exit accounting must not undo a replacement installed by a hot reload.
// The replaced adapter's loop is cancelled by Reload (a deliberate stop), so its
// exit says nothing about the adapter that now owns the name.
func TestReplacedAdapterExitKeepsReplacementConnected(t *testing.T) {
t.Parallel()
old := &signaledAdapter{
fakeAdapter: &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "old"}},
exited: make(chan struct{}),
}
factory := &stubFactory{adpts: []adapter.PlatformAdapter{
&fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "new"}},
}}
b := newReloadBridge(t, factory)
b.Add(old)
b.Start()
if err := b.Reload("kook"); err != nil {
t.Fatalf("Reload: %v", err)
}
<-old.exited
// The exit accounting runs immediately after the read path returns; a short
// settle window catches it without relying on a specific interleaving.
settle := time.Now().Add(50 * time.Millisecond)
for time.Now().Before(settle) {
if !b.IsConnected("kook:conf0") {
t.Fatal("the replaced adapter's read-loop exit cleared the replacement's connected flag")
}
time.Sleep(time.Millisecond)
}
}
+18 -3
View File
@@ -34,10 +34,16 @@ type PluginSummary struct {
// carries that platform's configured accounts (masked, no raw tokens) so the
// panel can show and edit the account config; it is omitted when s.channels is
// not wired or the platform has no configured account.
//
// NeedsRescan/LastError (v0.39) report that the adapter needs the owner to
// re-authorize it (see rescanAware in admin_channels_rescan.go); both are omitted
// unless the adapter really asked for it.
type ChannelSummary struct {
Name string `json:"name"`
Connected bool `json:"connected"`
Accounts []AdminChannelAccount `json:"accounts,omitempty"`
Name string `json:"name"`
Connected bool `json:"connected"`
NeedsRescan bool `json:"needsRescan,omitempty"`
LastError string `json:"lastError,omitempty"`
Accounts []AdminChannelAccount `json:"accounts,omitempty"`
}
// policyRequest is the POST /api/v1/admin/plugins/policy body.
@@ -172,6 +178,15 @@ func (s *Server) adminListChannels(w http.ResponseWriter, r *http.Request) {
continue
}
sum := ChannelSummary{Name: platform, Connected: ch.IsConnected()}
// Optional rescan capability (weixin-iLink expired token, v0.39): reported
// only when the adapter really asks for a re-scan, so a healthy adapter
// keeps the fields omitted rather than shipping a fabricated value.
if ra, ok := asRescanAware(cap.Value); ok {
if required, reason := ra.RescanRequired(); required {
sum.NeedsRescan = true
sum.LastError = reason
}
}
if accts, ok := accountsByPlatform[platform]; ok && len(accts) > 0 {
sum.Accounts = enrichAccountsWithBot(accts, botByPlatform[platform])
}
@@ -0,0 +1,62 @@
package gateway
import "reflect"
// This file holds the optional "the adapter needs the owner to re-authorize it"
// capability surfaced by GET /api/v1/admin/channels (v0.39). It lives next to —
// not inside — the channel summary itself because the summary list is one of
// admin.go's larger handlers and the unwrap helper is reusable.
// rescanAware is the optional capability a channel adapter exposes when its
// session needs owner re-authorization that a reconnect cannot provide.
// *weixin_ilink.Adapter implements it (RescanRequired): iLink answering
// errcode=-14 means the bot token is dead and the owner must scan a fresh QR. The
// core adapter.PlatformAdapter contract is deliberately NOT extended, so adapters
// that do not need the capability stay untouched.
//
// A value is only ever reported when the adapter really produced it — a platform
// without the capability keeps the fields omitted rather than shipping a guess.
type rescanAware interface {
// RescanRequired reports whether a new scan-code login is required, and why
// (reason is "" when it is not).
RescanRequired() (required bool, reason string)
}
// asRescanAware reports whether a registered channel value exposes the rescan
// capability. The registration wraps an adapter in cmd/cyrene's namedAdapter — an
// unexported struct embedding adapter.PlatformAdapter — which promotes only the
// interface's methods, so the concrete adapter's extra capability is reachable
// only by unwrapping the embedded field. A direct adapter value (tests,
// single-account registrations) is accepted as-is.
func asRescanAware(v any) (rescanAware, bool) {
if ra, ok := v.(rescanAware); ok {
return ra, true
}
inner, ok := unwrapNamedAdapter(v)
if !ok {
return nil, false
}
ra, ok := inner.(rescanAware)
return ra, ok
}
// unwrapNamedAdapter returns the adapter a registration wrapper embeds in its
// exported PlatformAdapter field. asKookAdmin and asWeixinLoginAdmin perform the
// same unwrap for their own capability subsets and could share this helper.
func unwrapNamedAdapter(v any) (any, bool) {
rv := reflect.ValueOf(v)
if !rv.IsValid() {
return nil, false
}
if rv.Kind() == reflect.Ptr {
rv = rv.Elem()
}
if rv.Kind() != reflect.Struct {
return nil, false
}
f := rv.FieldByName("PlatformAdapter")
if !f.IsValid() || !f.CanInterface() {
return nil, false
}
return f.Interface(), true
}
@@ -0,0 +1,242 @@
package gateway
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
"git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/weixin_ilink"
)
// rescanStubAdapter is a channel adapter that also exposes the optional rescan
// capability, exactly as *weixin_ilink.Adapter does once iLink answers
// errcode=-14 (expired bot token, re-scan required).
type rescanStubAdapter struct {
name string
connected bool
rescan bool
reason string
connectErr error
}
func (s *rescanStubAdapter) Name() string { return s.name }
func (s *rescanStubAdapter) Config() map[string]any { return map[string]any{"bot_token": "configured"} }
func (s *rescanStubAdapter) Connect(context.Context) error { return s.connectErr }
func (s *rescanStubAdapter) Disconnect(context.Context) error { return nil }
func (s *rescanStubAdapter) IsConnected() bool { return s.connected }
func (s *rescanStubAdapter) HealthCheck(context.Context) error { return nil }
func (s *rescanStubAdapter) SendMessage(context.Context, adapter.UnifiedResponse) error {
return nil
}
func (s *rescanStubAdapter) SendProactive(context.Context, adapter.UnifiedMessage) error { return nil }
func (s *rescanStubAdapter) SetMessageHandler(adapter.MessageHandler) {}
func (s *rescanStubAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() }
func (s *rescanStubAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) {
return nil, nil
}
func (s *rescanStubAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) {
<-ctx.Done()
}
func (s *rescanStubAdapter) RescanRequired() (bool, string) { return s.rescan, s.reason }
// namedRescanAdapter mirrors cmd/cyrene's namedAdapter: an unexported wrapper
// embedding adapter.PlatformAdapter, which promotes only the interface's methods.
// The gateway therefore has to unwrap it to reach the adapter's rescan capability.
type namedRescanAdapter struct{ adapter.PlatformAdapter }
var (
_ adapter.PlatformAdapter = (*rescanStubAdapter)(nil)
_ sdk.Channel = namedRescanAdapter{}
)
// TestAdminChannelsReportsNeedsRescan asserts the panel can tell "needs a fresh
// scan" apart from plain "not connected": the expired adapter's row carries
// needsRescan + the reason, and a healthy adapter omits both rather than shipping
// a fabricated value.
func TestAdminChannelsReportsNeedsRescan(t *testing.T) {
t.Parallel()
const reason = "iLink getupdates 返回 errcode=-14(bot token 已过期)"
caps := capreg.New()
// Registered through the wrapper, the way the composition root registers it.
if err := caps.Register(capreg.Capability{
Name: "weixin-ilink",
Kind: capreg.KindChannel,
Value: namedRescanAdapter{PlatformAdapter: &rescanStubAdapter{name: "weixin-ilink", rescan: true, reason: reason}},
}); err != nil {
t.Fatalf("register weixin-ilink: %v", err)
}
if err := caps.Register(capreg.Capability{
Name: "telegram",
Kind: capreg.KindChannel,
Value: &rescanStubAdapter{name: "telegram", connected: true},
}); err != nil {
t.Fatalf("register telegram: %v", err)
}
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Registry: caps})
tok := signAny(t, jwt, "acct", auth.RoleOwner)
items, body := adminChannelsList(t, ts, tok)
if len(items) != 2 {
t.Fatalf("channels = %+v, want 2 rows", items)
}
byName := map[string]ChannelSummary{}
for _, it := range items {
byName[it.Name] = it
}
wx, ok := byName["weixin-ilink"]
if !ok {
t.Fatalf("no weixin-ilink row in %+v", items)
}
if wx.Connected {
t.Error("weixin-ilink connected = true, want false while the token is expired")
}
if !wx.NeedsRescan {
t.Error("weixin-ilink needsRescan = false, want true (expired token must ask for a re-scan)")
}
if wx.LastError != reason {
t.Errorf("weixin-ilink lastError = %q, want %q", wx.LastError, reason)
}
tg, ok := byName["telegram"]
if !ok {
t.Fatalf("no telegram row in %+v", items)
}
if !tg.Connected || tg.NeedsRescan || tg.LastError != "" {
t.Errorf("telegram = %+v, want a connected row without rescan fields", tg)
}
// The healthy channel must OMIT the fields (omitempty), not send defaults: the
// panel distinguishes "adapter did not report" from "adapter reported false".
var raw []map[string]json.RawMessage
if err := json.Unmarshal(decodeBody(t, body)["data"], &raw); err != nil {
t.Fatalf("decode raw items: %v (body=%s)", err, body)
}
for _, item := range raw {
if string(item["name"]) != `"telegram"` {
continue
}
if _, present := item["needsRescan"]; present {
t.Error("telegram row carries needsRescan although the adapter never asked for a re-scan")
}
if _, present := item["lastError"]; present {
t.Error("telegram row carries lastError although there is none")
}
}
// The reason is informational text, never a credential.
if strings.Contains(string(body), `"bot_token"`) {
t.Errorf("channel list leaked a bot_token field: %s", body)
}
}
// TestAdminChannelOpenExpiredSessionReturns409 asserts the open endpoint answers
// honestly for a session a reconnect cannot fix: 409 session_expired with the
// re-scan instruction, never a 200 that would make the panel show "connected".
func TestAdminChannelOpenExpiredSessionReturns409(t *testing.T) {
t.Parallel()
caps := capreg.New()
if err := caps.Register(capreg.Capability{
Name: "weixin-ilink",
Kind: capreg.KindChannel,
Value: &rescanStubAdapter{
name: "weixin-ilink", rescan: true,
reason: "iLink bot token 已过期",
connectErr: weixin_ilink.ErrSessionExpired,
},
}); err != nil {
t.Fatalf("register weixin-ilink: %v", err)
}
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Registry: caps})
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/channels/open",
map[string]any{"name": "weixin-ilink"}, tok)
if resp.StatusCode != http.StatusConflict {
t.Fatalf("open status = %d, want 409 (body=%s)", resp.StatusCode, body)
}
var eobj struct {
Code string `json:"code"`
Message string `json:"message"`
}
if err := json.Unmarshal(decodeBody(t, body)["error"], &eobj); err != nil {
t.Fatalf("decode error envelope: %v (body=%s)", err, body)
}
if eobj.Code != "session_expired" {
t.Fatalf("error code = %q, want session_expired", eobj.Code)
}
if !strings.Contains(eobj.Message, "扫码") {
t.Fatalf("error message = %q, want the re-scan instruction", eobj.Message)
}
}
// TestAdminChannelsRescanFromRealIlinkAdapter closes the loop across the layers:
// a REAL *weixin_ilink.Adapter that iLink answered errcode=-14 for, registered the
// way the composition root registers it (wrapped), must surface needsRescan +
// lastError in the admin channel list.
func TestAdminChannelsRescanFromRealIlinkAdapter(t *testing.T) {
t.Parallel()
ilog := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
if r.URL.Path != "/ilink/bot/getupdates" {
w.WriteHeader(http.StatusNotFound)
return
}
_, _ = w.Write([]byte(`{"errcode":-14}`))
}))
defer ilog.Close()
ad, err := weixin_ilink.New(weixin_ilink.Config{
BaseURL: ilog.URL, BotToken: "tok", PollInterval: time.Millisecond,
}, ilog.Client(), nil)
if err != nil {
t.Fatalf("new weixin adapter: %v", err)
}
if err := ad.Connect(context.Background()); err != nil {
t.Fatalf("connect before expiry: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go ad.ReadMessages(ctx, func(adapter.UnifiedMessage) {})
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if required, _ := ad.RescanRequired(); required {
break
}
time.Sleep(time.Millisecond)
}
if required, _ := ad.RescanRequired(); !required {
t.Fatal("real adapter never reached the rescan-required state")
}
caps := capreg.New()
if err := caps.Register(capreg.Capability{
Name: "weixin-ilink",
Kind: capreg.KindChannel,
Value: namedRescanAdapter{PlatformAdapter: ad},
}); err != nil {
t.Fatalf("register weixin-ilink: %v", err)
}
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Registry: caps})
tok := signAny(t, jwt, "acct", auth.RoleOwner)
items, _ := adminChannelsList(t, ts, tok)
if len(items) != 1 {
t.Fatalf("channels = %+v, want one weixin-ilink row", items)
}
if items[0].Connected {
t.Error("connected = true although iLink rejected the token")
}
if !items[0].NeedsRescan {
t.Error("needsRescan = false, want true for an expired iLink session")
}
if items[0].LastError == "" {
t.Error("lastError is empty, want the re-scan reason")
}
}
@@ -236,6 +236,15 @@ func (s *Server) adminChannelOpen(w http.ResponseWriter, r *http.Request) {
}
if err := ch.Connect(r.Context()); err != nil {
s.auditClaims(r, "channel.open", "error")
// A state a plain reconnect cannot fix must not read as a generic failure:
// the weixin-iLink bot token was rejected by iLink (errcode=-14), so the
// owner needs a fresh scan-code login. Answer 409 with that instruction
// instead of pretending the channel opened (the panel marks it connected).
if errors.Is(err, weixin_ilink.ErrSessionExpired) {
writeError(w, apperr.Wrap(apperr.ErrConflict, "session_expired",
"该渠道会话已过期(需要重新扫码登录):请用「扫码登录」重新绑定后再连接"))
return
}
// weixin-iLink with no token yet must not hang nor read as a credential
// error: the panel routes the owner to the out-of-band 扫码登录 flow.
if errors.Is(err, weixin_ilink.ErrNeedsLogin) {