diff --git a/plugins/channels/weixin_ilink/adapter.go b/plugins/channels/weixin_ilink/adapter.go new file mode 100644 index 0000000..55b6739 --- /dev/null +++ b/plugins/channels/weixin_ilink/adapter.go @@ -0,0 +1,429 @@ +// Package weixin_ilink implements the WeChat ClawBot iLink channel adapter +// for Cyrene v2 (DevDocs/design/08-渠道接入.md §4). It adapts Tencent's iLink +// Bot API into the normalized internal/channel/adapter contract. +// +// The adapter covers the weixin-ilink surface: scan-code login, HTTP long +// polling with a persistent get_updates_buf cursor, per-peer context_token +// tracking, AES-128-ECB media encrypt/decrypt with CDN download, and the 24h +// proactive-push window. Media CDN references are captured lazily — the poll +// loop never downloads attachments, so a large file cannot stall message +// delivery (design 03 §8, item 6). +// +// All network I/O goes through an injected *http.Client and honors the passed +// context.Context, so tests can substitute an httptest.Server and cancel the +// read loop without leaking goroutines or timers. +package weixin_ilink + +import ( + "context" + "crypto/rand" + "encoding/base64" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "strings" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +const ( + // source labels every log record emitted by this package so failures can be + // traced back to the owning module (constraint 02 §5). + source = "plugins/channels/weixin_ilink" + + // platformName is the Name() value of the adapter. + platformName = "weixin-ilink" + + // defaultBaseURL and defaultCDNBaseURL come from the protocol summary. + defaultBaseURL = "https://ilinkai.weixin.qq.com" + defaultCDNBaseURL = "https://novac2c.cdn.weixin.qq.com/c2c" + + // defaultLongPollTimeout exceeds the ~35s server hold so the HTTP client + // does not abort a normal long-poll in flight. + defaultLongPollTimeout = 45 * time.Second + // defaultLoginTimeout bounds the scan-code confirmation wait. + defaultLoginTimeout = 8 * time.Minute + + // defaultProactiveWindow is what design/08 §4 calls the iLink 24h window. + defaultProactiveWindow = 24 * time.Hour + + // defaultMaxMediaSize caps a downloaded attachment at 50 MiB. + defaultMaxMediaSize int64 = 50 << 20 + + // botType is hard-coded in the official iLink source and is passed + // verbatim to get_bot_qrcode. + botType = 3 + // channelVersion is the base_info.channel_version sent on getupdates. + channelVersion = "1.0.2" + // errSessionExpired is the iLink errcode for an expired bot token. + errSessionExpired = -14 +) + +// Sentinel errors returned by this adapter, usable with errors.Is. +var ( + // ErrInvalidConfig is returned when a required config field is empty. + ErrInvalidConfig = errors.New("weixin_ilink: invalid config") + // ErrLoginTimeout is returned when the QR code is not confirmed in time. + ErrLoginTimeout = errors.New("weixin_ilink: qr login timed out") + // ErrSessionExpired is returned when iLink reports errcode=-14. + ErrSessionExpired = errors.New("weixin_ilink: session expired, re-scan required") + // ErrNoContextToken is returned when a send lacks the peer's context_token. + ErrNoContextToken = errors.New("weixin_ilink: no context_token for peer") + // ErrProactiveWindow is returned when a proactive push is outside 24h. + ErrProactiveWindow = errors.New("weixin_ilink: proactive push outside 24h window") + // ErrMediaTooLarge is returned when a download exceeds MaxMediaSize. + ErrMediaTooLarge = errors.New("weixin_ilink: media exceeds size limit") +) + +// Config holds the weixin-ilink adapter settings. It is sourced from +// environment/config by the channel layer and can be round-tripped through +// Config() and adapter.DecodeConfig. +type Config struct { + // BaseURL is the iLink API root. + BaseURL string `json:"base_url"` + // CDNBaseURL is the media CDN root used for downloads and uploads. + CDNBaseURL string `json:"cdn_base_url"` + // BotToken is the bearer token obtained from scan login. A non-empty token + // means Connect skips the QR flow. + BotToken string `json:"bot_token"` + // BotID is the ilink_bot_id from the login confirmation. + BotID string `json:"bot_id"` + // OwnerID is the ilink_user_id of the bound owner account. + OwnerID string `json:"owner_id"` + // Cursor is the persisted get_updates_buf. + Cursor string `json:"cursor"` + // LongPollTimeout is the per-request timeout for getupdates. + LongPollTimeout time.Duration `json:"long_poll_timeout"` + // LoginTimeout bounds how long the scan-code flow waits for confirmation. + LoginTimeout time.Duration `json:"login_timeout"` + // LoginInterval is how often the QR status is polled during login. + LoginInterval time.Duration `json:"login_interval"` + // PollInterval is the gap between long polls; zero means poll immediately. + PollInterval time.Duration `json:"poll_interval"` + // ProactiveWindow is how long after the user's last message a proactive + // push is allowed. + ProactiveWindow time.Duration `json:"proactive_window"` + // MaxMediaSize caps a downloaded attachment in bytes. + MaxMediaSize int64 `json:"max_media_size"` +} + +// withDefaults fills zero-valued fields with protocol defaults. BotID/OwnerID +// may legitimately be empty, so they are left untouched. +func (c Config) withDefaults() Config { + if c.BaseURL == "" { + c.BaseURL = defaultBaseURL + } + if c.CDNBaseURL == "" { + c.CDNBaseURL = defaultCDNBaseURL + } + if c.LongPollTimeout <= 0 { + c.LongPollTimeout = defaultLongPollTimeout + } + if c.LoginTimeout <= 0 { + c.LoginTimeout = defaultLoginTimeout + } + if c.LoginInterval <= 0 { + c.LoginInterval = loginPollInterval + } + if c.PollInterval < 0 { + c.PollInterval = 0 + } + if c.ProactiveWindow <= 0 { + c.ProactiveWindow = defaultProactiveWindow + } + if c.MaxMediaSize <= 0 { + c.MaxMediaSize = defaultMaxMediaSize + } + return c +} + +// Adapter is the PlatformAdapter implementation for WeChat ClawBot iLink. It is +// safe for concurrent use: reads and writes of the session flag, cursor, and +// per-peer token/timestamp maps are guarded by a single RWMutex. +type Adapter struct { + cfg Config + httpClient *http.Client + log *slog.Logger + clock func() time.Time + + mu sync.RWMutex + connected bool + qrcode string + contexts map[string]string + lastMsgAt map[string]time.Time + handler adapter.MessageHandler +} + +// New builds an Adapter. client may be nil to use http.DefaultClient; log may be +// nil to fall back to the project logger at info level. It returns +// ErrInvalidConfig when the base URL is blank after defaults. +func New(cfg Config, client *http.Client, log *slog.Logger) (*Adapter, error) { + cfg = cfg.withDefaults() + if !strings.HasPrefix(cfg.BaseURL, "http://") && !strings.HasPrefix(cfg.BaseURL, "https://") { + return nil, fmt.Errorf("%w: base url must be http(s): %q", ErrInvalidConfig, cfg.BaseURL) + } + if client == nil { + client = http.DefaultClient + } + if log == nil { + log = logger.New(logger.Options{}) + } + log = logger.WithSource(log, source) + + return &Adapter{ + cfg: cfg, + httpClient: client, + log: log, + clock: time.Now, + contexts: map[string]string{}, + lastMsgAt: map[string]time.Time{}, + }, nil +} + +// Name returns the platform identifier "weixin-ilink". +func (a *Adapter) Name() string { return platformName } + +// Config returns a snapshot of the current adapter configuration as a plain +// map, including the persisted cursor and login credentials. +func (a *Adapter) Config() map[string]any { + a.mu.RLock() + defer a.mu.RUnlock() + return map[string]any{ + "base_url": a.cfg.BaseURL, + "cdn_base_url": a.cfg.CDNBaseURL, + "bot_token": a.cfg.BotToken, + "bot_id": a.cfg.BotID, + "owner_id": a.cfg.OwnerID, + "cursor": a.cfg.Cursor, + "long_poll_timeout": a.cfg.LongPollTimeout, + "login_timeout": a.cfg.LoginTimeout, + "login_interval": a.cfg.LoginInterval, + "poll_interval": a.cfg.PollInterval, + "proactive_window": a.cfg.ProactiveWindow, + "max_media_size": a.cfg.MaxMediaSize, + } +} + +// Connect establishes the iLink session. It is idempotent. When a bot_token is +// already configured it is treated as a live session; otherwise it runs the +// scan-code login flow (see login.go). +func (a *Adapter) Connect(ctx context.Context) error { + a.mu.RLock() + connected := a.connected + token := a.cfg.BotToken + a.mu.RUnlock() + if connected && token != "" { + return nil + } + if token == "" { + if err := a.login(ctx); err != nil { + return fmt.Errorf("weixin_ilink: connect: %w", err) + } + } + a.mu.Lock() + a.connected = true + a.mu.Unlock() + a.log.Info("ilink session established", "bot_id", a.cfg.BotID) + return nil +} + +// Disconnect tears down the session. It is idempotent. +func (a *Adapter) Disconnect(ctx context.Context) error { + a.mu.Lock() + defer a.mu.Unlock() + if !a.connected { + return nil + } + a.connected = false + a.log.Info("ilink session closed") + return nil +} + +// IsConnected reports whether the session is currently live. +func (a *Adapter) IsConnected() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.connected +} + +// HealthCheck verifies the session is usable: it must be connected and carry a +// bot_token. It does not probe the network so callers can use it cheaply after +// a reconnect without incurring a request. +func (a *Adapter) HealthCheck(ctx context.Context) error { + a.mu.RLock() + defer a.mu.RUnlock() + if !a.connected || a.cfg.BotToken == "" { + return adapter.ErrNotConnected + } + return nil +} + +// Capabilities declares what the adapter supports. +func (a *Adapter) Capabilities() adapter.Capabilities { + caps := adapter.DefaultCapabilities() + caps.Proactive = true + return caps +} + +// ListChannels returns the bot account as a single direct channel. iLink group +// membership is not reliably enumerable, so only the bound account is listed. +func (a *Adapter) ListChannels(ctx context.Context) ([]adapter.ChannelInfo, error) { + a.mu.RLock() + botID := a.cfg.BotID + a.mu.RUnlock() + if botID == "" { + botID = "weixin-ilink" + } + return []adapter.ChannelInfo{{ + ChannelID: botID, + ChannelType: adapter.ChannelDirect, + ChannelName: "wechat", + }}, nil +} + +// SetMessageHandler installs the callback receiving push-style inbound +// messages. A nil fn clears it. When the handler is set, ReadMessages routes +// converted messages to it instead of the emit callback. +func (a *Adapter) SetMessageHandler(fn adapter.MessageHandler) { + a.mu.Lock() + defer a.mu.Unlock() + a.handler = fn +} + +func (a *Adapter) handlerLocked() adapter.MessageHandler { + a.mu.RLock() + defer a.mu.RUnlock() + return a.handler +} + +// --- concurrency helpers --- + +func (a *Adapter) cursor() string { + a.mu.RLock() + defer a.mu.RUnlock() + return a.cfg.Cursor +} + +func (a *Adapter) setCursor(c string) { + if c == "" { + return + } + a.mu.Lock() + a.cfg.Cursor = c + a.mu.Unlock() +} + +// recordInbound refreshes the per-peer context_token and last-message time +// (design/08 §8: every inbound message refreshes the token so proactive sends +// have a usable token within the window). +func (a *Adapter) recordInbound(token string, m adapter.UnifiedMessage) { + peer := m.SenderID + if m.IsGroup() { + peer = m.GroupID + } + if peer == "" { + return + } + a.mu.Lock() + a.contexts[peer] = token + a.lastMsgAt[peer] = m.Timestamp + a.mu.Unlock() +} + +// contextFor returns the stored context_token for a peer. +func (a *Adapter) contextFor(peer string) string { + a.mu.RLock() + defer a.mu.RUnlock() + return a.contexts[peer] +} + +// recipient resolves the destination peer and group flag for an outbound +// message: group id for group chats, else the sender id, else the channel id. +func recipient(m adapter.UnifiedMessage) (peer string, group bool) { + if m.IsGroup() { + return m.GroupID, true + } + if m.SenderID != "" { + return m.SenderID, false + } + return m.ChannelID, false +} + +// --- HTTP plumbing --- + +// authorizationHeader decorates req with the iLink common headers: JSON +// content type, the bearer token, the auth type, and a fresh random +// X-WECHAT-UIN replay guard. +func (a *Adapter) authorizationHeader(req *http.Request) { + req.Header.Set("Content-Type", "application/json") + req.Header.Set("AuthorizationType", "ilink_bot_token") + req.Header.Set("X-WECHAT-UIN", randomUIN()) + a.mu.RLock() + token := a.cfg.BotToken + a.mu.RUnlock() + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } +} + +// randomUIN returns a base64-encoded decimal uint32, minted fresh per request +// as iLink's anti-replay value. +func randomUIN() string { + var b [4]byte + _, _ = rand.Read(b[:]) + n := binary.BigEndian.Uint32(b[:]) + return base64.StdEncoding.EncodeToString([]byte(fmt.Sprint(n))) +} + +// doJSON performs an HTTP request and decodes the JSON response into out. It +// closes the response body and wraps network/decode errors. A nil out skips +// decoding (used for PUTs whose body is opaque bytes). +func (a *Adapter) doJSON(ctx context.Context, method, rawURL string, body io.Reader, out any) error { + req, err := http.NewRequestWithContext(ctx, method, rawURL, body) + if err != nil { + return fmt.Errorf("weixin_ilink: build %s %s: %w", method, rawURL, err) + } + a.authorizationHeader(req) + resp, err := a.httpClient.Do(req) + if err != nil { + return fmt.Errorf("weixin_ilink: %s %s: %w", method, rawURL, err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + msg, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return fmt.Errorf("weixin_ilink: %s %s: status %d: %s", + method, rawURL, resp.StatusCode, strings.TrimSpace(string(msg))) + } + if out == nil { + if _, err := io.Copy(io.Discard, resp.Body); err != nil { + return fmt.Errorf("weixin_ilink: drain %s %s: %w", method, rawURL, err) + } + return nil + } + if err := json.NewDecoder(resp.Body).Decode(out); err != nil { + return fmt.Errorf("weixin_ilink: decode %s %s: %w", method, rawURL, err) + } + return nil +} + +// apiURL joins a path with the configured base URL. +func (a *Adapter) apiURL(path string) string { + return strings.TrimRight(a.cfg.BaseURL, "/") + path +} + +// cdnURL joins a path with the configured CDN base URL, escaping raw query +// strings that some CDN references carry. +func (a *Adapter) cdnURL(raw string) string { + if u, err := url.Parse(raw); err == nil && u.IsAbs() { + return raw + } + return strings.TrimRight(a.cfg.CDNBaseURL, "/") + "/" + strings.TrimLeft(raw, "/") +} diff --git a/plugins/channels/weixin_ilink/adapter_test.go b/plugins/channels/weixin_ilink/adapter_test.go new file mode 100644 index 0000000..fabd908 --- /dev/null +++ b/plugins/channels/weixin_ilink/adapter_test.go @@ -0,0 +1,184 @@ +package weixin_ilink + +import ( + "context" + "errors" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestAdapter_Name_And_DefaultCapabilities(t *testing.T) { + t.Parallel() + ad, err := New(Config{BaseURL: "https://ilinkai.weixin.qq.com"}, nil, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + if got := ad.Name(); got != "weixin-ilink" { + t.Fatalf("Name() = %q, want weixin-ilink", got) + } + caps := ad.Capabilities() + if !caps.Proactive { + t.Error("Capabilities().Proactive = false, want true (iLink has a 24h window)") + } + if !caps.Image { + t.Error("Capabilities().Image = false, want true") + } + if got := caps.Normalize().MaxMessageLength; got <= 0 { + t.Fatalf("MaxMessageLength = %d, want > 0", got) + } +} + +func TestAdapter_New_RejectsBadBaseURL(t *testing.T) { + t.Parallel() + _, err := New(Config{BaseURL: "not-a-url"}, nil, nil) + if !errors.Is(err, ErrInvalidConfig) { + t.Fatalf("err = %v, want ErrInvalidConfig", err) + } +} + +func TestAdapter_Config_ReturnsSnapshot(t *testing.T) { + t.Parallel() + ad, err := New(Config{BaseURL: "https://ilinkai.weixin.qq.com", BotID: "bot-1"}, nil, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + cfg := ad.Config() + if cfg["bot_id"] != "bot-1" { + t.Fatalf("Config()[bot_id] = %v, want bot-1", cfg["bot_id"]) + } + if cfg["base_url"] != "https://ilinkai.weixin.qq.com" { + t.Fatalf("Config()[base_url] = %v", cfg["base_url"]) + } + // Round-tripping through DecodeConfig must preserve the typed config. + var back Config + if err := adapter.DecodeConfig(cfg, &back); err != nil { + t.Fatalf("DecodeConfig: %v", err) + } + if back.BaseURL != "https://ilinkai.weixin.qq.com" || back.BotID != "bot-1" { + t.Fatalf("decoded config = %+v, want base_url/bot_id preserved", back) + } +} + +func TestAdapter_Connect_WithToken_IsIdempotent(t *testing.T) { + t.Parallel() + ad, err := New(Config{BaseURL: "https://ilinkai.weixin.qq.com", BotToken: "tok-1"}, nil, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + if ad.IsConnected() { + t.Fatal("IsConnected() = true before Connect") + } + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if !ad.IsConnected() { + t.Fatal("IsConnected() = false after Connect") + } + // Second call is a no-op success. + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("second Connect: %v", err) + } +} + +func TestAdapter_HealthCheck(t *testing.T) { + t.Parallel() + t.Run("disconnected", func(t *testing.T) { + t.Parallel() + ad, _ := New(Config{BaseURL: "https://ilinkai.weixin.qq.com"}, nil, nil) + if err := ad.HealthCheck(context.Background()); !errors.Is(err, adapter.ErrNotConnected) { + t.Fatalf("HealthCheck = %v, want ErrNotConnected", err) + } + }) + t.Run("connected_with_token", func(t *testing.T) { + t.Parallel() + ad, _ := New(Config{BaseURL: "https://ilinkai.weixin.qq.com", BotToken: "tok-1"}, nil, nil) + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if err := ad.HealthCheck(context.Background()); err != nil { + t.Fatalf("HealthCheck = %v, want nil", err) + } + }) +} + +func TestAdapter_Disconnect_IsIdempotent(t *testing.T) { + t.Parallel() + ad, _ := New(Config{BaseURL: "https://ilinkai.weixin.qq.com", BotToken: "tok-1"}, nil, nil) + _ = ad.Connect(context.Background()) + if err := ad.Disconnect(context.Background()); err != nil { + t.Fatalf("Disconnect: %v", err) + } + if ad.IsConnected() { + t.Fatal("IsConnected() = true after Disconnect") + } + if err := ad.Disconnect(context.Background()); err != nil { + t.Fatalf("second Disconnect: %v", err) + } +} + +func TestAdapter_ListChannels(t *testing.T) { + t.Parallel() + ad, _ := New(Config{BaseURL: "https://ilinkai.weixin.qq.com", BotID: "bot-9"}, nil, nil) + chans, err := ad.ListChannels(context.Background()) + if err != nil { + t.Fatalf("ListChannels: %v", err) + } + if len(chans) != 1 { + t.Fatalf("ListChannels() = %d channels, want 1", len(chans)) + } + if chans[0].ChannelID != "bot-9" || chans[0].ChannelType != adapter.ChannelDirect { + t.Fatalf("channel = %+v, want {bot-9 direct}", chans[0]) + } +} + +func TestAdapter_SetMessageHandler_And_RecordInbound(t *testing.T) { + t.Parallel() + ad, _ := New(Config{BaseURL: "https://ilinkai.weixin.qq.com"}, nil, nil) + + called := false + ad.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { called = true }) + if ad.handlerLocked() == nil { + t.Fatal("handlerLocked() = nil after SetMessageHandler") + } + ad.SetMessageHandler(nil) + if ad.handlerLocked() != nil { + t.Fatal("handlerLocked() != nil after clearing") + } + if called { + t.Fatal("handler called before any message") + } + + // recordInbound must refresh the per-peer token and last-message time. + ad.recordInbound("tok-abc", adapter.UnifiedMessage{ + SenderID: "user-1", + Timestamp: ad.clock(), + }) + if got := ad.contextFor("user-1"); got != "tok-abc" { + t.Fatalf("contextFor(user-1) = %q, want tok-abc", got) + } +} + +func TestAdapter_Recipient_ResolvesGroupAndSender(t *testing.T) { + t.Parallel() + tests := []struct { + name string + msg adapter.UnifiedMessage + wantPeer string + wantGroup bool + }{ + {name: "direct sender", msg: adapter.UnifiedMessage{SenderID: "u1"}, wantPeer: "u1"}, + {name: "group id", msg: adapter.UnifiedMessage{ChannelType: adapter.ChannelGroup, GroupID: "g1"}, wantPeer: "g1", wantGroup: true}, + {name: "default channel id", msg: adapter.UnifiedMessage{ChannelID: "c1"}, wantPeer: "c1"}, + } + for _, tc := range tests { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + peer, group := recipient(tc.msg) + if peer != tc.wantPeer || group != tc.wantGroup { + t.Fatalf("recipient() = (%q,%v), want (%q,%v)", peer, group, tc.wantPeer, tc.wantGroup) + } + }) + } +} diff --git a/plugins/channels/weixin_ilink/aes.go b/plugins/channels/weixin_ilink/aes.go new file mode 100644 index 0000000..2d2a2eb --- /dev/null +++ b/plugins/channels/weixin_ilink/aes.go @@ -0,0 +1,79 @@ +package weixin_ilink + +import ( + "crypto/aes" + "errors" + "fmt" +) + +// ErrBadCiphertext is returned when ciphertext is not a whole number of AES +// blocks or its padding is malformed. +var ErrBadCiphertext = errors.New("weixin_ilink: bad ciphertext or padding") + +// pkcs7Pad appends PKCS#7 padding to src so its length becomes a multiple of +// blockSize. Every appended byte holds the pad length, so it self-describes +// and Unpad can strip it unambiguously. +func pkcs7Pad(src []byte, blockSize int) []byte { + n := blockSize - len(src)%blockSize + out := make([]byte, len(src)+n) + copy(out, src) + for i := len(src); i < len(out); i++ { + out[i] = byte(n) + } + return out +} + +// pkcs7Unpad strips PKCS#7 padding from src. It returns ErrBadCiphertext when +// the length is not a block multiple or the final pad byte is inconsistent. +func pkcs7Unpad(src []byte, blockSize int) ([]byte, error) { + if len(src) == 0 || len(src)%blockSize != 0 { + return nil, ErrBadCiphertext + } + n := int(src[len(src)-1]) + if n < 1 || n > blockSize { + return nil, ErrBadCiphertext + } + for _, b := range src[len(src)-n:] { + if int(b) != n { + return nil, ErrBadCiphertext + } + } + return src[:len(src)-n], nil +} + +// aesECBEncrypt encrypts src under AES with the supplied key using ECB mode. +// Go's crypto/cipher package does not ship ECB, so the mode is implemented +// here: src is padded to a block multiple and every block is encrypted +// independently with the same key. ECB is deterministic and not semantically +// secure; it is required here only because the iLink CDN protocol mandates it +// for media payloads. +func aesECBEncrypt(key, src []byte) ([]byte, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: new aes cipher: %w", err) + } + padded := pkcs7Pad(src, block.BlockSize()) + out := make([]byte, len(padded)) + for i := 0; i < len(padded); i += block.BlockSize() { + block.Encrypt(out[i:i+block.BlockSize()], padded[i:i+block.BlockSize()]) + } + return out, nil +} + +// aesECBDecrypt decrypts src under AES with the supplied key and strips the +// PKCS#7 padding. ECB has no IV, so the first block decrypts directly with the +// key. It returns ErrBadCiphertext for malformed input. +func aesECBDecrypt(key, src []byte) ([]byte, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: new aes cipher: %w", err) + } + if len(src) == 0 || len(src)%block.BlockSize() != 0 { + return nil, ErrBadCiphertext + } + out := make([]byte, len(src)) + for i := 0; i < len(src); i += block.BlockSize() { + block.Decrypt(out[i:i+block.BlockSize()], src[i:i+block.BlockSize()]) + } + return pkcs7Unpad(out, block.BlockSize()) +} diff --git a/plugins/channels/weixin_ilink/aes_test.go b/plugins/channels/weixin_ilink/aes_test.go new file mode 100644 index 0000000..6c88ff8 --- /dev/null +++ b/plugins/channels/weixin_ilink/aes_test.go @@ -0,0 +1,109 @@ +package weixin_ilink + +import ( + "bytes" + "crypto/aes" + "errors" + "strconv" + "testing" +) + +func TestAESECB_RoundTrip_AllBlockBoundaries(t *testing.T) { + t.Parallel() + key := bytes.Repeat([]byte{0x42}, aes.BlockSize) + // Exercise len 0 through just past two blocks (padding edge cases). + for n := 0; n <= 2*aes.BlockSize+5; n++ { + n := n + t.Run("len="+strconv.Itoa(n), func(t *testing.T) { + t.Parallel() + pt := bytes.Repeat([]byte{0xAB}, n) + ct, err := aesECBEncrypt(key, pt) + if err != nil { + t.Fatalf("aesECBEncrypt: %v", err) + } + got, err := aesECBDecrypt(key, ct) + if err != nil { + t.Fatalf("aesECBDecrypt: %v", err) + } + if !bytes.Equal(got, pt) { + t.Fatalf("round trip mismatch: got %x want %x", got, pt) + } + }) + } +} + +func TestAESECB_EncryptPadsToBlockSize(t *testing.T) { + t.Parallel() + key := bytes.Repeat([]byte{1}, aes.BlockSize) + ct, err := aesECBEncrypt(key, []byte("abc")) // 3 bytes -> 16 after pad + if err != nil { + t.Fatalf("aesECBEncrypt: %v", err) + } + if len(ct) != aes.BlockSize { + t.Fatalf("ciphertext = %d bytes, want %d", len(ct), aes.BlockSize) + } + // Full block input -> one pad block appended. + ct2, err := aesECBEncrypt(key, bytes.Repeat([]byte{0}, aes.BlockSize)) + if err != nil { + t.Fatalf("aesECBEncrypt: %v", err) + } + if len(ct2) != 2*aes.BlockSize { + t.Fatalf("full-block ciphertext = %d bytes, want %d", len(ct2), 2*aes.BlockSize) + } +} + +func TestAESECB_DecryptRejectsBadLength(t *testing.T) { + t.Parallel() + key := bytes.Repeat([]byte{2}, aes.BlockSize) + cases := []int{0, 5, aes.BlockSize - 1, aes.BlockSize + 1} + for _, n := range cases { + n := n + t.Run("len="+strconv.Itoa(n), func(t *testing.T) { + t.Parallel() + _, err := aesECBDecrypt(key, make([]byte, n)) + if !errors.Is(err, ErrBadCiphertext) { + t.Fatalf("err = %v, want ErrBadCiphertext", err) + } + }) + } +} + +func TestAESECB_DecryptRejectsBadPadding(t *testing.T) { + t.Parallel() + key := bytes.Repeat([]byte{3}, aes.BlockSize) + // Valid ciphertext for one block with a correct pad (0x10) ... + good, err := aesECBEncrypt(key, bytes.Repeat([]byte{4}, aes.BlockSize-1)) + if err != nil { + t.Fatalf("aesECBEncrypt: %v", err) + } + // ... then tamper the final pad byte so the padding check must fail. + bad := append([]byte(nil), good...) + bad[len(bad)-1] ^= 0xFF + _, err = aesECBDecrypt(key, bad) + if !errors.Is(err, ErrBadCiphertext) { + t.Fatalf("err = %v, want ErrBadCiphertext", err) + } +} + +func TestAESECB_KeySizesAccepted(t *testing.T) { + t.Parallel() + // AES-128 requires a 16-byte key. Verify out-of-range keys are rejected. + if _, err := aesECBEncrypt(bytes.Repeat([]byte{1}, aes.BlockSize-1), []byte("x")); err == nil { + t.Fatal("short key: expected error, got nil") + } + if _, err := aesECBEncrypt(bytes.Repeat([]byte{1}, aes.BlockSize+1), []byte("x")); err == nil { + t.Fatal("long key: expected error, got nil") + } +} + +func TestPKCS7_UnpadRejectsInvalid(t *testing.T) { + t.Parallel() + // More pad bytes than a block. + if _, err := pkcs7Unpad([]byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 17}, 16); !errors.Is(err, ErrBadCiphertext) { + t.Fatalf("oversize pad: err = %v, want ErrBadCiphertext", err) + } + // Zero pad byte is never valid. + if _, err := pkcs7Unpad([]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 0}, 16); !errors.Is(err, ErrBadCiphertext) { + t.Fatalf("zero pad: err = %v, want ErrBadCiphertext", err) + } +} diff --git a/plugins/channels/weixin_ilink/fake_test.go b/plugins/channels/weixin_ilink/fake_test.go new file mode 100644 index 0000000..2cffbb4 --- /dev/null +++ b/plugins/channels/weixin_ilink/fake_test.go @@ -0,0 +1,197 @@ +package weixin_ilink + +import ( + "context" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// handlerResult is the decision a test's fake handler returns for a request. +type handlerResult struct { + status int + body []byte + ctype string +} + +// reqRecorder captures the requests a fake server receives so tests can assert +// that the adapter sent the expected cursor, bearer token, payload, etc. +type reqRecorder struct { + mu sync.Mutex + reqs []recordedReq +} + +// recordedReq is a normalized snapshot of an inbound HTTP request. +type recordedReq struct { + method string + path string + query string + body []byte + auth string + authType string + uin string +} + +func (r *reqRecorder) add(req recordedReq) { + r.mu.Lock() + defer r.mu.Unlock() + r.reqs = append(r.reqs, req) +} + +func (r *reqRecorder) all() []recordedReq { + r.mu.Lock() + defer r.mu.Unlock() + return append([]recordedReq(nil), r.reqs...) +} + +func (r *reqRecorder) reset() { + r.mu.Lock() + defer r.mu.Unlock() + r.reqs = nil +} + +// find returns requests matching pred. +func (r *reqRecorder) find(pred func(recordedReq) bool) []recordedReq { + r.mu.Lock() + defer r.mu.Unlock() + var out []recordedReq + for _, q := range r.reqs { + if pred(q) { + out = append(out, q) + } + } + return out +} + +// bodyJSON decodes a request body into a plain map, returning nil on failure. +func (q recordedReq) bodyJSON() map[string]any { + var m map[string]any + if err := json.Unmarshal(q.body, &m); err != nil { + return nil + } + return m +} + +// jsonBody marshals v into a JSON byte slice. +func jsonBody(v any) []byte { + b, _ := json.Marshal(v) + return b +} + +// inlineJSON builds a json.RawMessage that inlines when placed in an []any, so +// the fake server can embed pre-serialized message objects into a msgs array. +func inlineJSON(v any) json.RawMessage { return json.RawMessage(jsonBody(v)) } + +// encodeToken builds a base64 bot_token whose trailing bytes are a known key, so +// deriveInboundKey yields a deterministic AES key (design/08 §3.3). +func encodeToken(key []byte) string { + return base64.StdEncoding.EncodeToString(append([]byte("prefix!"), key...)) +} + +// newJSONServer starts an httptest server whose handler records each request +// and answers via fn. The server is closed on test cleanup. +func newJSONServer(t *testing.T, rec *reqRecorder, fn func(r recordedReq) handlerResult) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + req := recordedReq{ + method: r.Method, + path: r.URL.Path, + query: r.URL.RawQuery, + body: body, + auth: r.Header.Get("Authorization"), + authType: r.Header.Get("AuthorizationType"), + uin: r.Header.Get("X-WECHAT-UIN"), + } + if rec != nil { + rec.add(req) + } + res := fn(req) + if res.ctype == "" { + res.ctype = "application/json" + } + if res.status == 0 { + res.status = http.StatusOK + } + w.Header().Set("Content-Type", res.ctype) + w.WriteHeader(res.status) + _, _ = w.Write(res.body) + })) + t.Cleanup(srv.Close) + return srv +} + +// newAdapter builds an Adapter wired to a test server. The caller controls the +// Config (BaseURL overridden from the server when left blank). +func newAdapter(t *testing.T, cfg Config, srv *httptest.Server) *Adapter { + t.Helper() + if srv != nil && cfg.BaseURL == "" { + cfg.BaseURL = srv.URL + } + var client *http.Client + if srv != nil { + client = srv.Client() + } + ad, err := New(cfg, client, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + return ad +} + +// newConnectedAdapter builds a connected Adapter carrying a bot token. +func newConnectedAdapter(t *testing.T, cfg Config, srv *httptest.Server) *Adapter { + t.Helper() + ad := newAdapter(t, cfg, srv) + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + return ad +} + +// conformance guards that the production adapter satisfies the contract at +// compile time. +var _ adapter.PlatformAdapter = (*Adapter)(nil) + +// waitFor polls fn until it returns true or the timeout elapses, failing the +// test on timeout. +func waitFor(t *testing.T, fn func() bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if fn() { + return + } + time.Sleep(2 * time.Millisecond) + } + t.Fatal("condition not met within timeout") +} + +// getupdatesReqs returns the recorded getupdates requests. +func getupdatesReqs(rec *reqRecorder) []recordedReq { + return rec.find(func(q recordedReq) bool { return q.path == "/ilink/bot/getupdates" }) +} + +// onceUpdates serves first (a raw getupdates body) exactly once, then empty +// polls. Read-loop tests use it so a repeated message is not double-emitted. +func onceUpdates(t *testing.T, rec *reqRecorder, first []byte) *httptest.Server { + t.Helper() + calls := 0 + return newJSONServer(t, rec, func(req recordedReq) handlerResult { + if req.path != "/ilink/bot/getupdates" { + return handlerResult{status: 404} + } + calls++ + if calls == 1 { + return handlerResult{body: first} + } + return handlerResult{body: jsonBody(map[string]any{"msgs": []any{}, "get_updates_buf": "c2"})} + }) +} diff --git a/plugins/channels/weixin_ilink/login.go b/plugins/channels/weixin_ilink/login.go new file mode 100644 index 0000000..8395bac --- /dev/null +++ b/plugins/channels/weixin_ilink/login.go @@ -0,0 +1,109 @@ +package weixin_ilink + +import ( + "context" + "fmt" + "net/http" + "net/url" + "strconv" + "time" +) + +// loginPollInterval is the default how often Connect polls the QR confirmation +// status. It is the documented ~1s cadence; the Config.LoginInterval field +// overrides it (used by tests to keep the flow short). +const loginPollInterval = 1 * time.Second + +// qrcodeResp is the get_bot_qrcode response. +type qrcodeResp struct { + Qrcode string `json:"qrcode"` + QrcodeImgContent string `json:"qrcode_img_content"` +} + +// qrcodeStatusResp is the get_qrcode_status response. The token and ids are +// only populated once status is "confirmed". +type qrcodeStatusResp struct { + Status string `json:"status"` + BotToken string `json:"bot_token"` + BotID string `json:"ilink_bot_id"` + UserID string `json:"ilink_user_id"` + BaseURL string `json:"baseurl"` +} + +// login runs the one-time scan-code binding (design/08 §4): request a QR code, +// then poll get_qrcode_status until the user confirms. On confirmation it +// persists bot_token, bot_id, owner_id and any returned baseurl, so later +// requests carry the Bearer token. +func (a *Adapter) login(ctx context.Context) error { + loginCtx, cancel := context.WithTimeout(ctx, a.cfg.LoginTimeout) + defer cancel() + + qr, err := a.fetchQRCode(loginCtx) + if err != nil { + if loginCtx.Err() != nil { + return fmt.Errorf("%w: %w", ErrLoginTimeout, loginCtx.Err()) + } + return fmt.Errorf("weixin_ilink: fetch qrcode: %w", err) + } + + a.mu.Lock() + a.qrcode = qr.Qrcode + a.mu.Unlock() + // Log only the renderable URL, never the raw qrcode (it is a secret). + a.log.Info("scan the qr code to bind the bot", "url", qr.QrcodeImgContent) + + ticker := time.NewTicker(a.cfg.LoginInterval) + defer ticker.Stop() + + for { + status, err := a.pollQRStatus(loginCtx, qr.Qrcode) + if err != nil { + // A deadline/cancel during the poll means the overall window + // elapsed; surface ErrLoginTimeout rather than the raw transport + // error. + if loginCtx.Err() != nil { + return fmt.Errorf("%w: %w", ErrLoginTimeout, loginCtx.Err()) + } + return fmt.Errorf("weixin_ilink: poll qr status: %w", err) + } + if status.Status == "confirmed" && status.BotToken != "" { + a.mu.Lock() + a.cfg.BotToken = status.BotToken + a.cfg.BotID = status.BotID + a.cfg.OwnerID = status.UserID + if status.BaseURL != "" { + a.cfg.BaseURL = status.BaseURL + } + a.mu.Unlock() + return nil + } + select { + case <-loginCtx.Done(): + return fmt.Errorf("%w: %w", ErrLoginTimeout, loginCtx.Err()) + case <-ticker.C: + } + } +} + +// fetchQRCode requests a new bot login QR code. +func (a *Adapter) fetchQRCode(ctx context.Context) (qrcodeResp, error) { + path := a.apiURL("/ilink/bot/get_bot_qrcode?bot_type=" + strconv.Itoa(botType)) + var out qrcodeResp + if err := a.doJSON(ctx, http.MethodGet, path, nil, &out); err != nil { + return qrcodeResp{}, err + } + if out.Qrcode == "" { + return qrcodeResp{}, fmt.Errorf("weixin_ilink: get_bot_qrcode returned empty qrcode") + } + return out, nil +} + +// pollQRStatus checks whether a QR code has been confirmed. +func (a *Adapter) pollQRStatus(ctx context.Context, qrcode string) (qrcodeStatusResp, error) { + path := a.apiURL("/ilink/bot/get_qrcode_status?qrcode=" + url.QueryEscape(qrcode)) + var out qrcodeStatusResp + if err := a.doJSON(ctx, http.MethodGet, path, nil, &out); err != nil { + return qrcodeStatusResp{}, err + } + return out, nil +} diff --git a/plugins/channels/weixin_ilink/login_test.go b/plugins/channels/weixin_ilink/login_test.go new file mode 100644 index 0000000..0670e03 --- /dev/null +++ b/plugins/channels/weixin_ilink/login_test.go @@ -0,0 +1,146 @@ +package weixin_ilink + +import ( + "context" + "errors" + "testing" + "time" +) + +// TestConnect_ScanLogin_StatusTransition drives the QR status flow +// wait → scaned → confirmed and asserts the credentials are stored. +func TestConnect_ScanLogin_StatusTransition(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + statusCalls := 0 + srv := newJSONServer(t, rec, func(req recordedReq) handlerResult { + switch req.path { + case "/ilink/bot/get_bot_qrcode": + return handlerResult{body: jsonBody(map[string]any{ + "qrcode": "qr-1", "qrcode_img_content": "https://img/qr", + })} + case "/ilink/bot/get_qrcode_status": + statusCalls++ + switch statusCalls { + case 1: + return handlerResult{body: jsonBody(map[string]any{"status": "wait"})} + case 2: + return handlerResult{body: jsonBody(map[string]any{"status": "scaned"})} + default: + return handlerResult{body: jsonBody(map[string]any{ + "status": "confirmed", + "bot_token": "tok-x", + "ilink_bot_id": "bot-x", + "ilink_user_id": "o-1", + "baseurl": "https://ilinkai.weixin.qq.com", + })} + } + } + return handlerResult{status: 404} + }) + + cfg := Config{BaseURL: srv.URL, LoginInterval: time.Millisecond, LoginTimeout: 5 * time.Second} + ad := newAdapter(t, cfg, srv) + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if !ad.IsConnected() { + t.Fatal("IsConnected() = false after scan login") + } + if got := ad.Config()["bot_token"]; got != "tok-x" { + t.Fatalf("bot_token = %v, want tok-x", got) + } + if got := ad.Config()["bot_id"]; got != "bot-x" { + t.Fatalf("bot_id = %v, want bot-x", got) + } + + reqs := rec.all() + if len(reqs) < 3 { + t.Fatalf("expected >=3 requests, got %d", len(reqs)) + } + // Confirm the qrcode id is echoed in the status poll query. + statusReqs := rec.find(func(q recordedReq) bool { + return q.path == "/ilink/bot/get_qrcode_status" + }) + if len(statusReqs) == 0 { + t.Fatal("no get_qrcode_status requests recorded") + } + if len(statusReqs) > 0 { + for i, q := range statusReqs { + if q.query != "qrcode=qr-1" { + t.Fatalf("status poll #%d query = %q, want qrcode=qr-1", i, q.query) + } + if q.authType != "ilink_bot_token" { + t.Fatalf("status poll #%d AuthorizationType = %q, want ilink_bot_token", i, q.authType) + } + } + } +} + +// TestConnect_ScanLogin_Timeout verifies the login aborts with ErrLoginTimeout +// when the QR is never confirmed. +func TestConnect_ScanLogin_Timeout(t *testing.T) { + t.Parallel() + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.path == "/ilink/bot/get_bot_qrcode" { + return handlerResult{body: jsonBody(map[string]any{"qrcode": "qr-t"})} + } + return handlerResult{body: jsonBody(map[string]any{"status": "wait"})} + }) + cfg := Config{BaseURL: srv.URL, LoginInterval: time.Millisecond, LoginTimeout: 25 * time.Millisecond} + ad := newAdapter(t, cfg, srv) + err := ad.Connect(context.Background()) + if !errors.Is(err, ErrLoginTimeout) { + t.Fatalf("err = %v, want ErrLoginTimeout", err) + } + if ad.IsConnected() { + t.Fatal("IsConnected() = true after failed login") + } +} + +// TestConnect_ScanLogin_EmptyQRCode verifies a qrcode-less response fails. +func TestConnect_ScanLogin_EmptyQRCode(t *testing.T) { + t.Parallel() + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.path == "/ilink/bot/get_bot_qrcode" { + return handlerResult{body: jsonBody(map[string]any{"qrcode": ""})} + } + return handlerResult{status: 404} + }) + ad := newAdapter(t, Config{BaseURL: srv.URL}, srv) + if err := ad.Connect(context.Background()); err == nil { + t.Fatal("Connect() = nil, want error for empty qrcode") + } +} + +// TestConnect_ScanLogin_ServerError verifies a non-2xx qrcode response fails. +func TestConnect_ScanLogin_ServerError(t *testing.T) { + t.Parallel() + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + return handlerResult{status: 500, body: []byte("boom")} + }) + ad := newAdapter(t, Config{BaseURL: srv.URL}, srv) + if err := ad.Connect(context.Background()); err == nil { + t.Fatal("Connect() = nil, want error on 500") + } +} + +// TestConnect_WithToken_SkipsQR verifies Connect does not touch the network +// when a bot token is already configured. +func TestConnect_WithToken_SkipsQR(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + srv := newJSONServer(t, rec, func(req recordedReq) handlerResult { + return handlerResult{status: 500, body: []byte("should not be called")} + }) + ad := newAdapter(t, Config{BaseURL: srv.URL, BotToken: "tok-pre"}, srv) + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if got := rec.all(); len(got) != 0 { + t.Fatalf("expected no network calls, got %d", len(got)) + } + if !ad.IsConnected() { + t.Fatal("IsConnected() = false with a pre-set token") + } +} diff --git a/plugins/channels/weixin_ilink/media.go b/plugins/channels/weixin_ilink/media.go new file mode 100644 index 0000000..98e8eda --- /dev/null +++ b/plugins/channels/weixin_ilink/media.go @@ -0,0 +1,96 @@ +package weixin_ilink + +import ( + "context" + "crypto/aes" + "encoding/base64" + "errors" + "fmt" + "io" + "net/http" + "strings" +) + +// ErrKeyTooShort is returned when the inbound derivation cannot yield an AES +// key of the required block size. +var ErrKeyTooShort = errors.New("weixin_ilink: derived key shorter than a block") + +// deriveInboundKey derives the AES-128 key for an inbound media payload. Per +// design/08 §3.3 the key is extracted from the bot_token: base64-decode the +// token and take the trailing 16 bytes. This is marked ⚠️ in the reference, so +// DownloadMediaWithKey lets callers supply an explicit key when the platform +// carries one per item instead. +func deriveInboundKey(botToken string) ([]byte, error) { + token := strings.TrimSpace(botToken) + if token == "" { + return nil, ErrInvalidConfig + } + b, err := base64.StdEncoding.DecodeString(token) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: decode bot_token for key: %w", err) + } + if len(b) < aes.BlockSize { + return nil, ErrKeyTooShort + } + return b[len(b)-aes.BlockSize:], nil +} + +// DownloadMedia downloads an inbound CDN reference and returns the decrypted +// plaintext using the key derived from the configured bot_token. It fails with +// ErrMediaTooLarge when the ciphertext exceeds MaxMediaSize. +func (a *Adapter) DownloadMedia(ctx context.Context, cdnURL string) ([]byte, error) { + key, err := deriveInboundKey(a.cfg.BotToken) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: derive inbound key: %w", err) + } + return a.downloadWithKey(ctx, cdnURL, key) +} + +// DownloadMediaWithKey is DownloadMedia with an explicit per-item AES key, +// used when the protocol delivers the key alongside the media reference. +func (a *Adapter) DownloadMediaWithKey(ctx context.Context, cdnURL string, key []byte) ([]byte, error) { + if len(key) != aes.BlockSize { + return nil, fmt.Errorf("weixin_ilink: key must be %d bytes: %w", aes.BlockSize, ErrKeyTooShort) + } + return a.downloadWithKey(ctx, cdnURL, key) +} + +// downloadWithKey fetches the CDN bytes and removes the AES-ECB layer. +func (a *Adapter) downloadWithKey(ctx context.Context, cdnURL string, key []byte) ([]byte, error) { + ciphertext, err := a.fetchBody(ctx, a.cdnURL(cdnURL)) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: fetch media: %w", err) + } + plaintext, err := aesECBDecrypt(key, ciphertext) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: decrypt media: %w", err) + } + return plaintext, nil +} + +// fetchBody GETs rawURL and returns the body, refusing anything over +// MaxMediaSize so a single huge file cannot exhaust memory. +func (a *Adapter) fetchBody(ctx context.Context, rawURL string) ([]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: build media request: %w", err) + } + a.authorizationHeader(req) + resp, err := a.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: download media: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("weixin_ilink: download media: status %d", resp.StatusCode) + } + limit := a.cfg.MaxMediaSize + 1 + data, err := io.ReadAll(io.LimitReader(resp.Body, limit)) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: read media: %w", err) + } + if int64(len(data)) > a.cfg.MaxMediaSize { + return nil, ErrMediaTooLarge + } + return data, nil +} diff --git a/plugins/channels/weixin_ilink/media_test.go b/plugins/channels/weixin_ilink/media_test.go new file mode 100644 index 0000000..fb4812b --- /dev/null +++ b/plugins/channels/weixin_ilink/media_test.go @@ -0,0 +1,145 @@ +package weixin_ilink + +import ( + "bytes" + "context" + "crypto/aes" + "errors" + "testing" +) + +func TestDeriveInboundKey(t *testing.T) { + t.Parallel() + key := bytes.Repeat([]byte{0x7F}, aes.BlockSize) + token := encodeToken(key) + + got, err := deriveInboundKey(token) + if err != nil { + t.Fatalf("deriveInboundKey: %v", err) + } + if !bytes.Equal(got, key) { + t.Fatalf("key = %x, want %x", got, key) + } +} + +func TestDeriveInboundKey_Errors(t *testing.T) { + t.Parallel() + t.Run("empty token", func(t *testing.T) { + t.Parallel() + if _, err := deriveInboundKey(""); !errors.Is(err, ErrInvalidConfig) { + t.Fatalf("err = %v, want ErrInvalidConfig", err) + } + }) + t.Run("not base64", func(t *testing.T) { + t.Parallel() + if _, err := deriveInboundKey("@not-base64@"); err == nil { + t.Fatal("err = nil, want decode error") + } + }) + t.Run("token too short", func(t *testing.T) { + t.Parallel() + if _, err := deriveInboundKey(encodeToken([]byte("short"))); !errors.Is(err, ErrKeyTooShort) { + t.Fatalf("err = %v, want ErrKeyTooShort", err) + } + }) +} + +func TestDownloadMedia_DecryptsFromTokenDerivedKey(t *testing.T) { + t.Parallel() + key := bytes.Repeat([]byte{0x11}, aes.BlockSize) + plain := []byte("the quick brown fox jumps over the lazy dog") + enc, err := aesECBEncrypt(key, plain) + if err != nil { + t.Fatalf("aesECBEncrypt: %v", err) + } + token := encodeToken(key) + + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.path == "/media/1" { + return handlerResult{ctype: "application/octet-stream", body: enc} + } + return handlerResult{status: 404} + }) + + ad := newAdapter(t, Config{BaseURL: srv.URL, CDNBaseURL: srv.URL, BotToken: token}, srv) + got, err := ad.DownloadMedia(context.Background(), srv.URL+"/media/1") + if err != nil { + t.Fatalf("DownloadMedia: %v", err) + } + if !bytes.Equal(got, plain) { + t.Fatalf("decrypted = %q, want %q", got, plain) + } +} + +func TestDownloadMediaWithKey_ExplicitKey(t *testing.T) { + t.Parallel() + key := bytes.Repeat([]byte{0x22}, aes.BlockSize) + plain := []byte("explicit key path") + enc, err := aesECBEncrypt(key, plain) + if err != nil { + t.Fatalf("aesECBEncrypt: %v", err) + } + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.path == "/media/2" { + return handlerResult{ctype: "application/octet-stream", body: enc} + } + return handlerResult{status: 404} + }) + ad := newAdapter(t, Config{BaseURL: srv.URL, CDNBaseURL: srv.URL, BotToken: "unused"}, srv) + got, err := ad.DownloadMediaWithKey(context.Background(), srv.URL+"/media/2", key) + if err != nil { + t.Fatalf("DownloadMediaWithKey: %v", err) + } + if !bytes.Equal(got, plain) { + t.Fatalf("decrypted = %q, want %q", got, plain) + } +} + +func TestDownloadMedia_RejectsOversize(t *testing.T) { + t.Parallel() + big := bytes.Repeat([]byte{0x01}, 256) + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.path == "/media/big" { + return handlerResult{ctype: "application/octet-stream", body: big} + } + return handlerResult{status: 404} + }) + ad := newAdapter(t, Config{ + BaseURL: srv.URL, CDNBaseURL: srv.URL, + BotToken: encodeToken(bytes.Repeat([]byte{1}, aes.BlockSize)), + MaxMediaSize: 64, + }, srv) + _, err := ad.DownloadMedia(context.Background(), srv.URL+"/media/big") + if !errors.Is(err, ErrMediaTooLarge) { + t.Fatalf("err = %v, want ErrMediaTooLarge", err) + } +} + +func TestDownloadMedia_RejectsDecryptFailure(t *testing.T) { + t.Parallel() + // Serve bytes that are not a valid AES block sequence. + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.path == "/media/bad" { + return handlerResult{ctype: "application/octet-stream", body: []byte("not-a-block")} + } + return handlerResult{status: 404} + }) + ad := newAdapter(t, Config{ + BaseURL: srv.URL, CDNBaseURL: srv.URL, + BotToken: encodeToken(bytes.Repeat([]byte{1}, aes.BlockSize)), + }, srv) + if _, err := ad.DownloadMedia(context.Background(), srv.URL+"/media/bad"); err == nil { + t.Fatal("err = nil, want decrypt failure") + } +} + +func TestCDNURL_JoinsRelativeAndPassesAbsolute(t *testing.T) { + t.Parallel() + ad, _ := New(Config{BaseURL: "https://api.example.com", CDNBaseURL: "https://cdn.example.com/c2c"}, nil, nil) + if got := ad.cdnURL("media/1.jpg"); got != "https://cdn.example.com/c2c/media/1.jpg" { + t.Fatalf("cdnURL(relative) = %q", got) + } + if got := ad.cdnURL("https://other.example.com/x"); got != "https://other.example.com/x" { + t.Fatalf("cdnURL(absolute) = %q", got) + } +} diff --git a/plugins/channels/weixin_ilink/poll.go b/plugins/channels/weixin_ilink/poll.go new file mode 100644 index 0000000..0ff70c4 --- /dev/null +++ b/plugins/channels/weixin_ilink/poll.go @@ -0,0 +1,378 @@ +package weixin_ilink + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +const ( + // msgTypeUser marks an inbound user message (bot messages are type 2). + msgTypeUser = 1 + // item type codes from the iLink reference (design 03 §1). + itemText = 1 + itemImage = 2 + itemVoice = 3 + itemFile = 4 + itemVideo = 5 + + // retryFastCount is how many consecutive failures use the fast retry. + retryFastCount = 3 + // retryFastDelay is the delay before the first failures. + retryFastDelay = 2 * time.Second + // retrySlowDelay is the backoff after retryFastCount consecutive failures. + retrySlowDelay = 30 * time.Second +) + +// getupdatesResp is the getupdates (long-poll) response envelope. +type getupdatesResp struct { + // Ret is the top-level business result; non-zero is a transient error. + Ret int `json:"ret"` + // Errcode is the platform error code; -14 means the session expired. + Errcode int `json:"errcode"` + // Msgs holds the raw inbound messages delivered in this poll. + Msgs []json.RawMessage `json:"msgs"` + // GetUpdatesBuf is the next Cursor to echo back. + GetUpdatesBuf string `json:"get_updates_buf"` + // LongPollTimeoutMs is the server's suggested hold time. + LongPollTimeoutMs int `json:"longpolling_timeout_ms"` +} + +// rawMsg is the parsed common fields of an inbound WeixinMessage. iLink media +// field names are only partially public, so item_list entries are decoded +// defensively in parseItem. +type rawMsg struct { + ClientID string `json:"client_id"` + FromUserID string `json:"from_user_id"` + ToUserID string `json:"to_user_id"` + MessageType int `json:"message_type"` + MessageState int `json:"message_state"` + ContextToken string `json:"context_token"` + GroupID string `json:"group_id"` + RoomID string `json:"room_id"` + ItemList []json.RawMessage `json:"item_list"` +} + +// ReadMessages runs the long-poll read loop until ctx is canceled. Each +// converted inbound message is delivered to the handler installed via +// SetMessageHandler when one is present, otherwise to emit. The loop is +// cooperative with ctx: a canceled context returns immediately and no timers +// outlive it. +func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) { + a.mu.RLock() + connected := a.connected + a.mu.RUnlock() + if !connected { + a.log.Warn("read loop started while disconnected") + return + } + + failures := 0 + for { + resp, err := a.fetchUpdates(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + // A long-poll timeout is the normal "no messages yet" path. + if errors.Is(err, context.DeadlineExceeded) { + continue + } + failures++ + a.log.Warn("getupdates failed", "err", err, "failures", failures) + if !a.wait(ctx, a.retryDelay(failures)) { + return + } + continue + } + + failures = 0 + if resp.Errcode == errSessionExpired { + a.log.Warn("session expired (errcode=-14); stop polling, re-scan required") + a.mu.Lock() + a.connected = false + a.mu.Unlock() + return + } + if resp.Ret != 0 { + failures++ + a.log.Warn("getupdates business error", "ret", resp.Ret) + if !a.wait(ctx, a.retryDelay(failures)) { + return + } + continue + } + a.setCursor(resp.GetUpdatesBuf) + + for _, raw := range resp.Msgs { + token, m := a.convertMessage(raw) + if m == nil { + continue + } + a.recordInbound(token, *m) + a.deliver(ctx, *m, emit) + } + if !a.wait(ctx, a.cfg.PollInterval) { + return + } + } +} + +// deliver routes a normalized message to the handler when installed, else to +// the emit callback (design/08 §1: push-style vs pull-style delivery). +func (a *Adapter) deliver(ctx context.Context, m adapter.UnifiedMessage, emit func(adapter.UnifiedMessage)) { + if h := a.handlerLocked(); h != nil { + h(ctx, m) + return + } + if emit != nil { + emit(m) + } +} + +// fetchUpdates performs one long poll. The request context is bounded by +// LongPollTimeout if the caller has not already set a tighter deadline. +func (a *Adapter) fetchUpdates(ctx context.Context) (*getupdatesResp, error) { + reqCtx, cancel := context.WithTimeout(ctx, a.cfg.LongPollTimeout) + defer cancel() + + body, err := json.Marshal(map[string]any{ + "get_updates_buf": a.cursor(), + "base_info": map[string]string{"channel_version": channelVersion}, + }) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: encode getupdates: %w", err) + } + var out getupdatesResp + if err := a.doJSON(reqCtx, http.MethodPost, a.apiURL("/ilink/bot/getupdates"), + bytes.NewReader(body), &out); err != nil { + return nil, fmt.Errorf("weixin_ilink: getupdates: %w", err) + } + return &out, nil +} + +// wait sleeps for d or returns false when ctx is canceled. A non-positive +// delay returns immediately. +func (a *Adapter) wait(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return true + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} + +// retryDelay grows from the fast retry to the slow backoff after enough +// consecutive failures (design/08 §5.4). +func (a *Adapter) retryDelay(failures int) time.Duration { + if failures <= retryFastCount { + return retryFastDelay + } + return retrySlowDelay +} + +// convertMessage maps a raw iLink message to a UnifiedMessage and returns the +// raw context_token so the caller can track it per peer. It returns nil when +// the message is not an inbound user message or cannot be parsed. +func (a *Adapter) convertMessage(raw json.RawMessage) (string, *adapter.UnifiedMessage) { + var rm rawMsg + if err := json.Unmarshal(raw, &rm); err != nil { + a.log.Warn("decode inbound message", "err", err) + return "", nil + } + if rm.MessageType != msgTypeUser { + return "", nil + } + + group := resolveGroup(&rm) + m := &adapter.UnifiedMessage{ + Platform: platformName, + SenderID: rm.FromUserID, + SenderName: rm.FromUserID, + MessageID: rm.ClientID, + Timestamp: a.clock(), + RawData: raw, + } + if group != "" { + m.ChannelType = adapter.ChannelGroup + m.GroupID = group + m.ChannelID = group + } else { + m.ChannelType = adapter.ChannelDirect + m.ChannelID = rm.FromUserID + } + m.Sender = adapter.Actor{ + Role: adapter.RoleUnknown, + UserID: rm.FromUserID, + DisplayName: rm.FromUserID, + IsGroup: group != "", + GroupID: group, + } + m.Content, m.ContentType, m.Media = a.convertItems(rm.ItemList) + return rm.ContextToken, m +} + +// convertItems drains the item list into text, a primary content type and a +// media slice. Text from text items is concatenated; voice items contribute +// their ASR transcript; image/file/video items are attached lazily as CDN +// references (never downloaded during the poll). +func (a *Adapter) convertItems(rawItems []json.RawMessage) (string, adapter.ContentType, []adapter.Media) { + var text strings.Builder + var media []adapter.Media + firstMedia := adapter.ContentType("") + + for _, raw := range rawItems { + item := parseItem(raw) + switch item.kind { + case itemText: + text.WriteString(item.text) + case itemVoice: + text.WriteString(item.text) + media = append(media, item.media) + firstMedia = adapter.ContentVoice + case itemImage: + media = append(media, item.media) + firstMedia = adapter.ContentImage + case itemFile, itemVideo: + media = append(media, item.media) + firstMedia = adapter.ContentFile + } + } + + content := strings.TrimRight(text.String(), "\n") + if firstMedia != "" { + return content, firstMedia, media + } + return content, adapter.ContentText, nil +} + +// resolveGroup returns the group identifier when the message carries one, +// otherwise "" (direct chat). WeChat group signalling is ⚠️ per design/08 §4, +// so several candidate fields are checked defensively. +func resolveGroup(rm *rawMsg) string { + if rm.GroupID != "" { + return rm.GroupID + } + return rm.RoomID +} + +// parsedItem is a normalized view of a single native item_list entry. +type parsedItem struct { + kind int + text string + media adapter.Media +} + +// parseItem decodes one item_list entry defensively. iLink media field names +// are not fully public, so a generic map is scanned for the CDN reference and +// metadata with several fallback keys. +func parseItem(raw json.RawMessage) parsedItem { + var m map[string]any + if err := json.Unmarshal(raw, &m); err != nil { + return parsedItem{} + } + kind := intOf(m["type"]) + switch kind { + case itemText: + return parsedItem{kind: itemText, text: textOf(m, "text_item")} + case itemVoice: + return parsedItem{kind: itemVoice, text: textOf(m, "voice_item"), media: mediaOf(m, adapter.MediaVoice)} + case itemImage: + return parsedItem{kind: itemImage, media: mediaOf(m, adapter.MediaImage)} + case itemFile: + return parsedItem{kind: itemFile, media: mediaOf(m, adapter.MediaFile)} + case itemVideo: + return parsedItem{kind: itemVideo, media: mediaOf(m, adapter.MediaVideo)} + default: + return parsedItem{kind: kind} + } +} + +// textOf reads the text payload from the named nested sub-object, falling back +// to a flat "text" key. +func textOf(m map[string]any, sub string) string { + if inner, ok := m[sub].(map[string]any); ok { + if s, ok := inner["text"].(string); ok { + return s + } + } + if s, ok := m["text"].(string); ok { + return s + } + return "" +} + +// mediaOf builds an adapter.Media from a media item, resolving the CDN URL and +// metadata from the flat item or a nested sub-object. +func mediaOf(m map[string]any, kind adapter.MediaKind) adapter.Media { + res := adapter.Media{Kind: kind} + res.URL = firstString(m, "url", "cdn_url", "download_url", "file_url", "cdn_ref") + res.Name = firstString(m, "file_name", "name") + res.Size = firstInt64(m, "size", "file_size") + res.MIMEType = firstString(m, "mime", "mime_type", "content_type") + if res.URL != "" { + return res + } + // Bounded scan of the common nested sub-objects for the CDN reference. + for _, sub := range []string{"image_item", "file_item", "video_item", "media_item", "cdn_item"} { + inner, ok := m[sub].(map[string]any) + if !ok { + continue + } + res.URL = firstString(inner, "url", "cdn_url", "download_url", "file_url", "cdn_ref") + if res.Name == "" { + res.Name = firstString(inner, "file_name", "name") + } + if res.Size == 0 { + res.Size = firstInt64(inner, "size", "file_size") + } + if res.MIMEType == "" { + res.MIMEType = firstString(inner, "mime", "mime_type", "content_type") + } + if res.URL != "" { + return res + } + } + return res +} + +// intOf coerces a JSON number to int, returning 0 for missing or non-numeric. +func intOf(v any) int { + f, ok := v.(float64) + if !ok { + return 0 + } + return int(f) +} + +// firstString returns the first non-empty string among keys. +func firstString(m map[string]any, keys ...string) string { + for _, k := range keys { + if s, ok := m[k].(string); ok && s != "" { + return s + } + } + return "" +} + +// firstInt64 returns the first non-zero integer among keys. +func firstInt64(m map[string]any, keys ...string) int64 { + for _, k := range keys { + if f, ok := m[k].(float64); ok && f != 0 { + return int64(f) + } + } + return 0 +} diff --git a/plugins/channels/weixin_ilink/poll_test.go b/plugins/channels/weixin_ilink/poll_test.go new file mode 100644 index 0000000..0af3461 --- /dev/null +++ b/plugins/channels/weixin_ilink/poll_test.go @@ -0,0 +1,279 @@ +package weixin_ilink + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// msgCollector collects emitted messages under a mutex. +type msgCollector struct { + mu sync.Mutex + msgs []adapter.UnifiedMessage +} + +func (c *msgCollector) add(m adapter.UnifiedMessage) { + c.mu.Lock() + defer c.mu.Unlock() + c.msgs = append(c.msgs, m) +} + +func (c *msgCollector) snapshot() []adapter.UnifiedMessage { + c.mu.Lock() + defer c.mu.Unlock() + return append([]adapter.UnifiedMessage(nil), c.msgs...) +} + +func (c *msgCollector) count() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.msgs) +} + +func textItem(t string) map[string]any { + return map[string]any{"type": 1, "text_item": map[string]any{"text": t}} +} + +func TestReadMessages_LongPoll_CursorAndContextToken(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + msg1, msg2 := longPollMessages() + + updatesCalls := 0 + srv := newJSONServer(t, rec, func(req recordedReq) handlerResult { + if req.path != "/ilink/bot/getupdates" { + return handlerResult{status: 404} + } + updatesCalls++ + if updatesCalls == 1 { + return handlerResult{body: jsonBody(map[string]any{ + "msgs": []any{msg1, msg2}, "get_updates_buf": "cursor-1", + })} + } + return handlerResult{body: jsonBody(map[string]any{ + "msgs": []any{}, "get_updates_buf": "cursor-2", + })} + }) + + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv) + + col := &msgCollector{} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + ad.ReadMessages(ctx, col.add) + }() + + waitFor(t, func() bool { return col.count() >= 2 }) + // Ensure the second long poll has run so the cursor echo is observable. + waitFor(t, func() bool { return len(getupdatesReqs(rec)) >= 2 }) + cancel() + waitDone(t, done) + + assertDecodedMessages(t, col.snapshot()) + assertCursorAndContext(t, ad, rec) +} + +func longPollMessages() (json.RawMessage, json.RawMessage) { + msg1 := inlineJSON(map[string]any{ + "client_id": "m-1", "from_user_id": "user-1", "to_user_id": "bot-1", + "message_type": 1, "message_state": 2, "context_token": "tok-a", + "item_list": []any{textItem("hi")}, + }) + msg2 := inlineJSON(map[string]any{ + "client_id": "m-2", "from_user_id": "user-2", "to_user_id": "bot-1", + "message_type": 1, "message_state": 2, "context_token": "tok-b", + "item_list": []any{map[string]any{"type": 2, "image_item": map[string]any{"url": "https://cdn/x.jpg"}}}, + }) + return msg1, msg2 +} + +func assertDecodedMessages(t *testing.T, msgs []adapter.UnifiedMessage) { + t.Helper() + if len(msgs) != 2 { + t.Fatalf("emitted %d messages, want 2", len(msgs)) + } + m := msgs[0] + if m.Platform != "weixin-ilink" || m.SenderID != "user-1" || m.ChannelID != "user-1" { + t.Fatalf("msg0 identity = platform %q sender %q channel %q", m.Platform, m.SenderID, m.ChannelID) + } + if m.ChannelType != adapter.ChannelDirect || m.ContentType != adapter.ContentText || m.Content != "hi" { + t.Fatalf("msg0 = channel %q content %q type %q", m.ChannelType, m.Content, m.ContentType) + } + m2 := msgs[1] + if m2.ContentType != adapter.ContentImage || len(m2.Media) != 1 { + t.Fatalf("msg1 = type %q media %d", m2.ContentType, len(m2.Media)) + } + if m2.Media[0].URL != "https://cdn/x.jpg" || m2.Media[0].Kind != adapter.MediaImage { + t.Fatalf("msg1 media = %+v", m2.Media[0]) + } + if m2.Timestamp.IsZero() { + t.Fatal("msg1 Timestamp is zero") + } +} + +func assertCursorAndContext(t *testing.T, ad *Adapter, rec *reqRecorder) { + t.Helper() + if got := ad.contextFor("user-1"); got != "tok-a" { + t.Fatalf("contextFor(user-1) = %q, want tok-a", got) + } + if got := ad.contextFor("user-2"); got != "tok-b" { + t.Fatalf("contextFor(user-2) = %q, want tok-b", got) + } + ups := getupdatesReqs(rec) + if len(ups) < 2 { + t.Fatalf("getupdates requests = %d, want >=2", len(ups)) + } + body2 := ups[1].bodyJSON() + if body2 == nil { + t.Fatal("second getupdates body is not JSON") + } + if got := body2["get_updates_buf"]; got != "cursor-1" { + t.Fatalf("second poll get_updates_buf = %v, want cursor-1", got) + } + if ups[0].auth != "Bearer tok" { + t.Fatalf("poll Authorization = %q, want Bearer tok", ups[0].auth) + } +} + +// waitDone waits for the read loop goroutine to exit or fails after a timeout. +func waitDone(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("ReadMessages did not return after ctx cancel") + } +} + +func TestReadMessages_SkipsNonUserMessages(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + srv := onceUpdates(t, rec, jsonBody(map[string]any{ + "msgs": []any{ + inlineJSON(map[string]any{"message_type": 1, "from_user_id": "u-1", "item_list": []any{textItem("yo")}}), + inlineJSON(map[string]any{"message_type": 2, "from_user_id": "bot-1", "item_list": []any{textItem("echo")}}), + }, + "get_updates_buf": "cursor-1", + })) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv) + + col := &msgCollector{} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + ad.ReadMessages(ctx, col.add) + }() + waitFor(t, func() bool { return col.count() >= 1 }) + cancel() + <-done + + if got := col.count(); got != 1 { + t.Fatalf("emitted %d messages, want 1 (bot echo skipped)", got) + } + if msgs := col.snapshot(); msgs[0].Content != "yo" { + t.Fatalf("content = %q, want yo", msgs[0].Content) + } +} + +func TestReadMessages_GroupMessage(t *testing.T) { + t.Parallel() + srv := onceUpdates(t, nil, jsonBody(map[string]any{ + "msgs": []any{inlineJSON(map[string]any{ + "message_type": 1, "from_user_id": "u-1", "group_id": "g-1", + "context_token": "tok-g", "item_list": []any{textItem("group msg")}, + })}, + "get_updates_buf": "c1", + })) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv) + + col := &msgCollector{} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + ad.ReadMessages(ctx, col.add) + }() + waitFor(t, func() bool { return col.count() >= 1 }) + cancel() + <-done + + m := col.snapshot()[0] + if m.ChannelType != adapter.ChannelGroup || m.GroupID != "g-1" || m.ChannelID != "g-1" { + t.Fatalf("group message = channel %q group %q channelid %q", m.ChannelType, m.GroupID, m.ChannelID) + } + if !m.Sender.IsGroup || m.Sender.GroupID != "g-1" { + t.Fatalf("sender group flags = %+v", m.Sender) + } + if m.IsGroup() != true { + t.Fatal("IsGroup() = false, want true") + } +} + +func TestReadMessages_SessionExpired(t *testing.T) { + t.Parallel() + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.path != "/ilink/bot/getupdates" { + return handlerResult{status: 404} + } + return handlerResult{body: jsonBody(map[string]any{"errcode": -14})} + }) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + ad.ReadMessages(ctx, func(m adapter.UnifiedMessage) {}) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + cancel() + t.Fatal("ReadMessages did not stop on session expired") + } + cancel() + if ad.IsConnected() { + t.Fatal("IsConnected() = true after session expired") + } +} + +func TestReadMessages_HandlerDelivery(t *testing.T) { + t.Parallel() + srv := onceUpdates(t, nil, jsonBody(map[string]any{ + "msgs": []any{inlineJSON(map[string]any{ + "message_type": 1, "from_user_id": "u-1", "context_token": "tok-h", + "item_list": []any{textItem("hello")}, + })}, + "get_updates_buf": "c1", + })) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv) + + handlerCalls := &msgCollector{} + ad.SetMessageHandler(func(_ context.Context, m adapter.UnifiedMessage) { handlerCalls.add(m) }) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + // emit is intentionally nil: the handler must receive the message. + ad.ReadMessages(ctx, nil) + }() + waitFor(t, func() bool { return handlerCalls.count() >= 1 }) + cancel() + <-done + + if got := handlerCalls.count(); got != 1 { + t.Fatalf("handler called %d times, want 1", got) + } + if m := handlerCalls.snapshot()[0]; m.Content != "hello" { + t.Fatalf("handler content = %q, want hello", m.Content) + } +} diff --git a/plugins/channels/weixin_ilink/send.go b/plugins/channels/weixin_ilink/send.go new file mode 100644 index 0000000..5648025 --- /dev/null +++ b/plugins/channels/weixin_ilink/send.go @@ -0,0 +1,247 @@ +package weixin_ilink + +import ( + "bytes" + "context" + "crypto/aes" + "crypto/rand" + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "strconv" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// item kind constants reuse the iLink item type codes from poll.go, so +// outbound media maps to the same numbers as inbound parsing. + +// sendResult is the sendmessage response envelope. +type sendResult struct { + Ret int `json:"ret"` + Errcode int `json:"errcode"` +} + +// SendMessage delivers a reply to the channel. Each message is sent to the +// peer resolved from the message (or resp.ReplyTo) using that peer's stored +// context_token, so the platform can associate the reply with the right +// conversation (design/08 §8). +func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) error { + if !a.IsConnected() { + return adapter.ErrNotConnected + } + for _, m := range resp.Messages { + peer, _ := recipient(m) + if peer == "" { + peer = resp.ReplyTo + } + if err := a.sendToOne(ctx, peer, m); err != nil { + return fmt.Errorf("weixin_ilink: send message: %w", err) + } + } + return nil +} + +// SendProactive pushes a message that was not triggered by a recent inbound. +// WeChat iLink only allows this inside the 24h window since the user's last +// message (design/08 §3-4): when the user has been silent longer, or has never +// messaged, the push is rejected with ErrProactiveWindow and Capabilities. +// Proactive is only true so callers know the capability exists. +func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error { + if !a.IsConnected() { + return adapter.ErrNotConnected + } + peer, _ := recipient(msg) + if peer == "" { + return ErrNoContextToken + } + + a.mu.RLock() + last, seen := a.lastMsgAt[peer] + window := a.cfg.ProactiveWindow + a.mu.RUnlock() + + if !seen || a.clock().Sub(last) > window { + return ErrProactiveWindow + } + if a.contextFor(peer) == "" { + return ErrNoContextToken + } + return a.sendToOne(ctx, peer, msg) +} + +// sendToOne sends one normalized message to a peer as a native iLink message, +// composing its item_list from any text plus uploaded media. +func (a *Adapter) sendToOne(ctx context.Context, peer string, m adapter.UnifiedMessage) error { + if peer == "" { + return ErrNoContextToken + } + ctxToken := a.contextFor(peer) + if ctxToken == "" { + return ErrNoContextToken + } + + items := make([]any, 0, 1+len(m.Media)) + if m.Content != "" { + items = append(items, nativeTextItem(m.Content)) + } + for _, md := range m.Media { + item, err := a.buildMediaItem(ctx, md) + if err != nil { + return fmt.Errorf("weixin_ilink: build media item for %q: %w", md.Kind, err) + } + items = append(items, item) + } + if len(items) == 0 { + return nil + } + return a.sendNative(ctx, peer, ctxToken, items) +} + +// buildMediaItem encrypts media bytes, uploads them to the CDN and returns the +// native item referencing the uploaded object and its AES key. +func (a *Adapter) buildMediaItem(ctx context.Context, md adapter.Media) (any, error) { + plaintext := md.Data + if len(plaintext) == 0 && md.URL != "" { + data, err := a.fetchBody(ctx, md.URL) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: fetch media source: %w", err) + } + plaintext = data + } + + key := make([]byte, aes.BlockSize) + if _, err := rand.Read(key); err != nil { + return nil, fmt.Errorf("weixin_ilink: generate aes key: %w", err) + } + encrypted, err := aesECBEncrypt(key, plaintext) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: encrypt media: %w", err) + } + uploadURL, err := a.getUploadURL(ctx) + if err != nil { + return nil, fmt.Errorf("weixin_ilink: get upload url: %w", err) + } + if err := a.putBytes(ctx, uploadURL, encrypted); err != nil { + return nil, fmt.Errorf("weixin_ilink: upload media: %w", err) + } + return nativeMediaItem(nativeMediaTypeFor(md.Kind), uploadURL, key, md), nil +} + +// sendNative posts a composed sendmessage payload. +func (a *Adapter) sendNative(ctx context.Context, peer, ctxToken string, items []any) error { + body, err := json.Marshal(map[string]any{ + "msg": map[string]any{ + "to_user_id": peer, + "client_id": newClientID(), + "message_type": 2, // bot message + "message_state": 2, // FINISH + "context_token": ctxToken, + "item_list": items, + }, + }) + if err != nil { + return fmt.Errorf("weixin_ilink: encode sendmessage: %w", err) + } + var out sendResult + if err := a.doJSON(ctx, http.MethodPost, a.apiURL("/ilink/bot/sendmessage"), + bytes.NewReader(body), &out); err != nil { + return fmt.Errorf("weixin_ilink: sendmessage: %w", err) + } + if out.Errcode == errSessionExpired { + return ErrSessionExpired + } + if out.Ret != 0 { + return fmt.Errorf("weixin_ilink: sendmessage ret=%d", out.Ret) + } + return nil +} + +// getUploadURL requests a pre-signed CDN upload URL. The response field names +// vary across server versions, so several candidates are tried. +func (a *Adapter) getUploadURL(ctx context.Context) (string, error) { + body, err := json.Marshal(map[string]any{}) + if err != nil { + return "", fmt.Errorf("weixin_ilink: encode getuploadurl: %w", err) + } + var raw map[string]any + if err := a.doJSON(ctx, http.MethodPost, a.apiURL("/ilink/bot/getuploadurl"), + bytes.NewReader(body), &raw); err != nil { + return "", fmt.Errorf("weixin_ilink: getuploadurl: %w", err) + } + u := firstString(raw, "upload_url", "url", "uploadUrl", "cdn_url") + if u == "" { + return "", fmt.Errorf("weixin_ilink: getuploadurl returned no url") + } + return u, nil +} + +// putBytes uploads encrypted media bytes to a pre-signed CDN URL. +func (a *Adapter) putBytes(ctx context.Context, uploadURL string, data []byte) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPut, uploadURL, bytes.NewReader(data)) + if err != nil { + return fmt.Errorf("weixin_ilink: build upload: %w", err) + } + req.Header.Set("Content-Type", "application/octet-stream") + resp, err := a.httpClient.Do(req) + if err != nil { + return fmt.Errorf("weixin_ilink: upload media: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("weixin_ilink: upload media: status %d", resp.StatusCode) + } + return nil +} + +// nativeTextItem builds a text item_list entry. +func nativeTextItem(text string) map[string]any { + return map[string]any{ + "type": itemText, + "text_item": map[string]any{"text": text}, + } +} + +// nativeMediaItem builds a media item_list entry that references the uploaded +// object. Both cdn_ref and url are set because the exact field is not public. +func nativeMediaItem(kind int, ref string, key []byte, md adapter.Media) map[string]any { + item := map[string]any{ + "type": kind, + "cdn_ref": ref, + "url": ref, + "aes_key": base64.StdEncoding.EncodeToString(key), + "file_name": md.Name, + "size": md.Size, + } + if md.MIMEType != "" { + item["mime"] = md.MIMEType + } + return item +} + +// nativeMediaTypeFor maps a normalized MediaKind to the iLink item type code. +func nativeMediaTypeFor(kind adapter.MediaKind) int { + switch kind { + case adapter.MediaImage: + return itemImage + case adapter.MediaVoice: + return itemVoice + case adapter.MediaVideo: + return itemVideo + default: + return itemFile + } +} + +// newClientID returns a random hex client id used to de-duplicate sends. On an +// (unlikely) crypto/rand failure it falls back to a time-based value. +func newClientID() string { + var b [12]byte + if _, err := rand.Read(b[:]); err == nil { + return hex.EncodeToString(b[:]) + } + return "c-" + strconv.FormatInt(time.Now().UnixNano(), 36) +} diff --git a/plugins/channels/weixin_ilink/send_test.go b/plugins/channels/weixin_ilink/send_test.go new file mode 100644 index 0000000..5a1f072 --- /dev/null +++ b/plugins/channels/weixin_ilink/send_test.go @@ -0,0 +1,229 @@ +package weixin_ilink + +import ( + "bytes" + "context" + "crypto/aes" + "encoding/base64" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// sendOKServer replies ret:0 to sendmessage, recording every request. +func sendOKServer(t *testing.T, rec *reqRecorder) *httptest.Server { + t.Helper() + return newJSONServer(t, rec, func(req recordedReq) handlerResult { + if req.method == http.MethodPost && req.path == "/ilink/bot/sendmessage" { + return handlerResult{body: jsonBody(map[string]any{"ret": 0})} + } + return handlerResult{status: 404} + }) +} + +func TestSendMessage_RequiresConnection(t *testing.T) { + t.Parallel() + ad, err := New(Config{BaseURL: "https://ilinkai.weixin.qq.com"}, nil, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + err = ad.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{SenderID: "u-1", Content: "hi"}}, + }) + if !errors.Is(err, adapter.ErrNotConnected) { + t.Fatalf("err = %v, want ErrNotConnected", err) + } +} + +func TestSendMessage_RequiresContextToken(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + srv := sendOKServer(t, rec) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, BotToken: "tok"}, srv) + + err := ad.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{SenderID: "u-1", Content: "hi"}}, + }) + if !errors.Is(err, ErrNoContextToken) { + t.Fatalf("err = %v, want ErrNoContextToken", err) + } +} + +func TestSendMessage_SendsTextReply(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + srv := sendOKServer(t, rec) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, BotToken: "tok"}, srv) + + ad.recordInbound("tok-1", adapter.UnifiedMessage{SenderID: "u-1", Timestamp: ad.clock()}) + + resp := adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{SenderID: "u-1", Content: "pong"}}, + ReplyTo: "anything", + } + if err := ad.SendMessage(context.Background(), resp); err != nil { + t.Fatalf("SendMessage: %v", err) + } + body := getSendBody(t, rec) + msg := body["msg"].(map[string]any) + if msg["to_user_id"] != "u-1" { + t.Fatalf("to_user_id = %v, want u-1", msg["to_user_id"]) + } + if msg["context_token"] != "tok-1" { + t.Fatalf("context_token = %v, want tok-1", msg["context_token"]) + } + item := msg["item_list"].([]any)[0].(map[string]any) + if typ, ok := item["type"].(float64); !ok || typ != 1 { + t.Fatalf("item type = %v, want 1", item["type"]) + } + if txt := item["text_item"].(map[string]any)["text"]; txt != "pong" { + t.Fatalf("text = %v, want pong", txt) + } + if msg["client_id"] == "" || msg["client_id"] == nil { + t.Fatal("client_id missing") + } +} + +func TestSendProactive_WithinWindow(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + srv := sendOKServer(t, rec) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, BotToken: "tok"}, srv) + + now := time.Date(2026, 9, 6, 12, 0, 0, 0, time.UTC) + ad.clock = func() time.Time { return now } + ad.recordInbound("tok-p", adapter.UnifiedMessage{SenderID: "u-1", Timestamp: now}) + + err := ad.SendProactive(context.Background(), adapter.UnifiedMessage{ + SenderID: "u-1", Content: "daily tip", + }) + if err != nil { + t.Fatalf("SendProactive within window: %v", err) + } + body := getSendBody(t, rec) + msg := body["msg"].(map[string]any) + if msg["to_user_id"] != "u-1" || msg["context_token"] != "tok-p" { + t.Fatalf("proactive send = to_user_id %v context %v", msg["to_user_id"], msg["context_token"]) + } +} + +func TestSendProactive_ExpiredWindow(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + srv := sendOKServer(t, rec) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, BotToken: "tok"}, srv) + + now := time.Date(2026, 9, 6, 12, 0, 0, 0, time.UTC) + ad.clock = func() time.Time { return now } + ad.recordInbound("tok-p", adapter.UnifiedMessage{SenderID: "u-1", Timestamp: now}) + // Advance beyond the 24h window. + now = now.Add(25 * time.Hour) + + err := ad.SendProactive(context.Background(), adapter.UnifiedMessage{ + SenderID: "u-1", Content: "too late", + }) + if !errors.Is(err, ErrProactiveWindow) { + t.Fatalf("err = %v, want ErrProactiveWindow", err) + } + if len(getupdatesReqs(rec)) != 0 { + t.Fatal("expired window must not send") + } +} + +func TestSendProactive_NoPriorInbound(t *testing.T) { + t.Parallel() + srv := sendOKServer(t, &reqRecorder{}) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, BotToken: "tok"}, srv) + err := ad.SendProactive(context.Background(), adapter.UnifiedMessage{ + SenderID: "u-1", Content: "hello", + }) + if !errors.Is(err, ErrProactiveWindow) { + t.Fatalf("err = %v, want ErrProactiveWindow (no prior inbound)", err) + } +} + +func TestSendMessage_OutboundMedia(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + var srv *httptest.Server + srv = newJSONServer(t, rec, func(req recordedReq) handlerResult { + switch { + case req.method == http.MethodPost && req.path == "/ilink/bot/getuploadurl": + return handlerResult{body: jsonBody(map[string]any{"upload_url": srv.URL + "/upload/1"})} + case req.method == http.MethodPut && req.path == "/upload/1": + return handlerResult{status: 200, body: []byte("ok")} + case req.method == http.MethodPost && req.path == "/ilink/bot/sendmessage": + return handlerResult{body: jsonBody(map[string]any{"ret": 0})} + default: + return handlerResult{status: 404} + } + }) + + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, CDNBaseURL: srv.URL, BotToken: "tok"}, srv) + ad.recordInbound("tok-m", adapter.UnifiedMessage{SenderID: "u-1", Timestamp: ad.clock()}) + + imageData := []byte("fake-jpeg-bytes") + err := ad.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ + SenderID: "u-1", + Media: []adapter.Media{{Kind: adapter.MediaImage, Data: imageData, Name: "a.jpg"}}, + }}, + }) + if err != nil { + t.Fatalf("SendMessage media: %v", err) + } + + body := getSendBody(t, rec) + msg := body["msg"].(map[string]any) + item := msg["item_list"].([]any)[0].(map[string]any) + if typ, ok := item["type"].(float64); !ok || typ != 2 { + t.Fatalf("media item type = %v, want 2", item["type"]) + } + if item["cdn_ref"] != srv.URL+"/upload/1" { + t.Fatalf("cdn_ref = %v, want upload url", item["cdn_ref"]) + } + aesKeyEncoded, _ := item["aes_key"].(string) + if aesKeyEncoded == "" { + t.Fatal("aes_key missing on media item") + } + key, err := base64.StdEncoding.DecodeString(aesKeyEncoded) + if err != nil { + t.Fatalf("decode aes_key: %v", err) + } + if len(key) != aes.BlockSize { + t.Fatalf("aes_key = %d bytes, want %d", len(key), aes.BlockSize) + } + + // The uploaded bytes (PUT body) must decrypt back to the original media. + puts := rec.find(func(q recordedReq) bool { return q.method == http.MethodPut }) + if len(puts) != 1 { + t.Fatalf("PUT requests = %d, want 1", len(puts)) + } + decrypted, err := aesECBDecrypt(key, puts[0].body) + if err != nil { + t.Fatalf("decrypt uploaded media: %v", err) + } + if !bytes.Equal(decrypted, imageData) { + t.Fatalf("uploaded media mismatch: got %q want %q", decrypted, imageData) + } +} + +// getSendBody returns the decoded body of the first sendmessage request. +func getSendBody(t *testing.T, rec *reqRecorder) map[string]any { + t.Helper() + reqs := rec.find(func(q recordedReq) bool { + return q.method == http.MethodPost && q.path == "/ilink/bot/sendmessage" + }) + if len(reqs) == 0 { + t.Fatal("no sendmessage request recorded") + } + body := reqs[0].bodyJSON() + if body == nil { + t.Fatal("sendmessage body is not JSON") + } + return body +}