diff --git a/cmd/imwsaudit/main.go b/cmd/imwsaudit/main.go new file mode 100644 index 0000000..7ca00c2 --- /dev/null +++ b/cmd/imwsaudit/main.go @@ -0,0 +1,1050 @@ +package main + +import ( + "bytes" + "crypto/tls" + "encoding/base64" + "encoding/json" + "flag" + "fmt" + "io" + "mime/multipart" + "net/http" + "net/url" + "os" + "sort" + "strings" + "time" + + "github.com/gorilla/websocket" +) + +type imResp struct { + Code int `json:"code"` + Reason string `json:"reason"` + Data map[string]any `json:"data"` +} + +type result struct { + ID string + Name string + Method string + Target string + Pass bool + Note string +} + +type audit struct { + baseURL string + wsURL string + apiKey string + tenantCode string + out string + client *http.Client + results []result + usernameA string + phoneA string + passA string + userA string + tokenA string + usernameB string + phoneB string + passB string + userB string + tokenB string + p2pTopic string + bobTopic string + groupTopic string + fileURL string + messageID string +} + +type wsClient struct { + name string + user string + token string + conn *websocket.Conn + pending []map[string]any +} + +func main() { + baseURL := flag.String("base", "https://api.118im.com", "HTTP base URL") + wsURL := flag.String("ws", "wss://api.118im.com/v0/channels", "WebSocket URL") + apiKey := flag.String("apikey", "AQAAAAABAAAoeOI7tA3HsYvdzDhYhZJy", "Tinode API key") + tenant := flag.String("tenant", "default", "tenant code") + out := flag.String("out", "docs/test/im_websocket_testenv_results_2026-08-21.md", "report path") + flag.Parse() + + a := &audit{ + baseURL: strings.TrimRight(*baseURL, "/"), + wsURL: *wsURL, + apiKey: *apiKey, + tenantCode: *tenant, + out: *out, + client: &http.Client{ + Timeout: 20 * time.Second, + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}, + }, + }, + } + a.run() + if err := os.WriteFile(a.out, []byte(a.report()), 0o644); err != nil { + panic(err) + } + fmt.Printf("wrote %s\n", a.out) +} + +func (a *audit) run() { + suffix := fmt.Sprintf("%08d", time.Now().UnixNano()%100000000) + nameSuffix := suffix[len(suffix)-6:] + a.usernameA, a.phoneA, a.passA = "wsa"+nameSuffix, "157"+suffix, "AuditPass123" + a.usernameB, a.phoneB, a.passB = "wsb"+nameSuffix, "158"+suffix, "AuditPass123" + + a.step("WS-01", "注册用户 A: HTTP 验证码 + WS acc", "WS", "/v0/channels", func() string { + user, token, err := a.register(a.usernameA, a.phoneA, a.passA) + if err != nil { + return "FAIL: " + err.Error() + } + a.userA, a.tokenA = user, token + return "user=" + user + " token=" + mask(token) + }) + a.step("WS-02", "注册用户 B: HTTP 验证码 + WS acc", "WS", "/v0/channels", func() string { + user, token, err := a.register(a.usernameB, a.phoneB, a.passB) + if err != nil { + return "FAIL: " + err.Error() + } + a.userB, a.tokenB = user, token + return "user=" + user + " token=" + mask(token) + }) + a.step("WS-03", "HTTP 建立好友关系生成 P2P Topic", "HTTP", "/v1/im/friend-requests", func() string { + reqID, err := a.createFriendship() + if err != nil { + return "FAIL: " + err.Error() + } + return "request_id=" + reqID + " topic=" + a.p2pTopic + }) + + var alice, bob *wsClient + a.step("WS-04", "A token login + sub me", "WS", "/v0/channels", func() string { + c, err := a.openLoggedIn("A", a.userA, a.tokenA) + if err != nil { + return "FAIL: " + err.Error() + } + alice = c + return "login user=" + a.userA + }) + a.step("WS-05", "B token login + sub me", "WS", "/v0/channels", func() string { + c, err := a.openLoggedIn("B", a.userB, a.tokenB) + if err != nil { + return "FAIL: " + err.Error() + } + bob = c + return "login user=" + a.userB + }) + defer func() { + if alice != nil && alice.conn != nil { + _ = alice.conn.Close() + } + if bob != nil && bob.conn != nil { + _ = bob.conn.Close() + } + }() + + a.step("WS-06", "A/B 订阅 P2P Topic", "WS", a.p2pTopic, func() string { + if alice == nil || bob == nil { + return "FAIL: websocket client missing" + } + if err := alice.sub(a.p2pTopic); err != nil { + return "FAIL: A sub: " + err.Error() + } + if err := bob.sub(a.p2pTopic); err != nil { + return "FAIL: B sub: " + err.Error() + } + return "topic=" + a.p2pTopic + }) + a.step("WS-07", "A pub 文本消息，B 收到 data", "WS", a.p2pTopic, func() string { + if alice == nil || bob == nil { + return "FAIL: websocket client missing" + } + seq, bobTopic, err := alice.pubAndPeerReceives(bob, a.p2pTopic, "WS 联调消息 "+suffix) + if err != nil { + return "FAIL: " + err.Error() + } + a.messageID = fmt.Sprintf("%d", seq) + a.bobTopic = bobTopic + return fmt.Sprintf("seq=%d bob_topic=%s", seq, bobTopic) + }) + a.step("WS-08", "B 发送 recv/read 回执", "WS", a.p2pTopic, func() string { + if alice == nil || bob == nil || a.messageID == "" || a.bobTopic == "" { + return "FAIL: websocket client or seq missing" + } + var seq int + _, _ = fmt.Sscanf(a.messageID, "%d", &seq) + if err := bob.note(a.bobTopic, "recv", seq); err != nil { + return "FAIL: recv: " + err.Error() + } + if err := bob.note(a.bobTopic, "read", seq); err != nil { + return "FAIL: read: " + err.Error() + } + info, err := alice.wait(func(msg map[string]any) bool { + i, ok := msg["info"].(map[string]any) + if !ok || asString(i["what"]) != "read" || asInt(i["seq"]) < seq { + return false + } + topic := asString(i["topic"]) + return topic == a.p2pTopic || topic == a.userB + }, 5*time.Second) + if err != nil { + return "FAIL: wait read info: " + err.Error() + } + i := info["info"].(map[string]any) + return fmt.Sprintf("what=%s seq=%d", asString(i["what"]), asInt(i["seq"])) + }) + + a.step("WS-09", "basic login + sub me", "WS", "/v0/channels", func() string { + c, err := a.openBasicLoggedIn("basic-a", a.usernameA, a.passA) + if err != nil { + return "FAIL: " + err.Error() + } + defer c.conn.Close() + return "basic login ok" + }) + a.step("WS-10", "重复 login 被拒绝", "WS", "/v0/channels", func() string { + if alice == nil { + return "FAIL: websocket client missing" + } + code, text, err := alice.sendCtrl("login-duplicate", map[string]any{"login": map[string]any{"id": "login-duplicate", "scheme": "token", "secret": a.tokenA}}, 5*time.Second) + if err != nil { + return "FAIL: " + err.Error() + } + if code == 409 || code == 400 { + return fmt.Sprintf("code=%d text=%s", code, text) + } + return fmt.Sprintf("FAIL: unexpected code=%d text=%s", code, text) + }) + a.step("WS-11", "订阅 fnd 搜索 Topic", "WS", "fnd", func() string { + if alice == nil { + return "FAIL: websocket client missing" + } + if err := alice.sub("fnd"); err != nil { + return "FAIL: " + err.Error() + } + return "sub fnd ok" + }) + a.step("WS-12", "get me desc/sub/tags/cred", "WS", "me", func() string { + if alice == nil { + return "FAIL: websocket client missing" + } + if _, err := alice.get("me", "desc sub tags cred", nil); err != nil { + return "FAIL: " + err.Error() + } + return "get me ok" + }) + a.step("WS-13", "get P2P desc/sub/data/del", "WS", a.p2pTopic, func() string { + if alice == nil { + return "FAIL: websocket client missing" + } + if _, err := alice.get(a.p2pTopic, "desc sub data del", map[string]any{"data": map[string]any{"since": 1, "limit": 20}, "del": map[string]any{"since": 1, "limit": 20}}); err != nil { + return "FAIL: " + err.Error() + } + return "get p2p ok" + }) + a.step("WS-14", "set P2P private 设置", "WS", a.p2pTopic, func() string { + if alice == nil { + return "FAIL: websocket client missing" + } + if err := alice.setPrivate(a.p2pTopic, map[string]any{"mute": false, "pinned": true, "source": "imwsaudit"}); err != nil { + return "FAIL: " + err.Error() + } + return "set private ok" + }) + a.step("WS-15", "pub 产品 x-im 基础头", "WS", a.p2pTopic, func() string { + if alice == nil || bob == nil { + return "FAIL: websocket client missing" + } + text := "WS 产品头消息 " + suffix + seq, topic, err := alice.pubWithHeadAndPeer(bob, a.p2pTopic, map[string]any{"mime": "text/plain", "x-im-schema": 1, "x-im-type": "text"}, text) + if err != nil { + return "FAIL: " + err.Error() + } + return fmt.Sprintf("seq=%d bob_topic=%s", seq, topic) + }) + a.step("WS-16", "pub 产品 x-im-client-mid 幂等头", "WS", a.p2pTopic, func() string { + if alice == nil || bob == nil { + return "FAIL: websocket client missing" + } + text := "WS 幂等头消息 " + suffix + seq, topic, err := alice.pubWithHeadAndPeer(bob, a.p2pTopic, map[string]any{"mime": "text/plain", "x-im-schema": 1, "x-im-type": "text", "x-im-client-mid": "mid-" + suffix}, text) + if err != nil { + return "FAIL: " + err.Error() + } + return fmt.Sprintf("seq=%d bob_topic=%s", seq, topic) + }) + a.step("WS-17", "note kp 输入中通知", "WS", a.bobTopic, func() string { + if alice == nil || bob == nil || a.bobTopic == "" { + return "FAIL: websocket client missing" + } + if err := bob.note(a.bobTopic, "kp", 0); err != nil { + return "FAIL: " + err.Error() + } + msg, err := alice.wait(func(msg map[string]any) bool { + i, ok := msg["info"].(map[string]any) + return ok && asString(i["what"]) == "kp" + }, 5*time.Second) + if err != nil { + return "FAIL: " + err.Error() + } + i := msg["info"].(map[string]any) + return "what=" + asString(i["what"]) + }) + a.step("WS-18", "del 当前用户消息视图", "WS", a.p2pTopic, func() string { + if alice == nil || a.messageID == "" { + return "FAIL: websocket client or seq missing" + } + var seq int + _, _ = fmt.Sscanf(a.messageID, "%d", &seq) + if err := alice.delMsg(a.p2pTopic, seq); err != nil { + return "FAIL: " + err.Error() + } + return fmt.Sprintf("deleted seq=%d", seq) + }) + a.step("WS-19", "leave P2P detach 后重新订阅", "WS", a.p2pTopic, func() string { + if alice == nil { + return "FAIL: websocket client missing" + } + if err := alice.leave(a.p2pTopic, false); err != nil { + return "FAIL: leave: " + err.Error() + } + if err := alice.sub(a.p2pTopic); err != nil { + return "FAIL: resub: " + err.Error() + } + return "leave/resub ok" + }) + a.step("WS-20", "HTTP 创建群供 WS 群协议测试", "HTTP", "/v1/im/groups", func() string { + topic, err := a.createGroup(suffix) + if err != nil { + return "FAIL: " + err.Error() + } + a.groupTopic = topic + return "topic=" + topic + }) + a.step("WS-21", "A/B 订阅群 Topic", "WS", a.groupTopic, func() string { + if alice == nil || bob == nil || a.groupTopic == "" { + return "FAIL: websocket client or group missing" + } + if err := alice.sub(a.groupTopic); err != nil { + return "FAIL: A sub group: " + err.Error() + } + if err := bob.sub(a.groupTopic); err != nil { + return "FAIL: B sub group: " + err.Error() + } + return "sub group ok" + }) + a.step("WS-22", "群 pub 文本广播", "WS", a.groupTopic, func() string { + if alice == nil || bob == nil || a.groupTopic == "" { + return "FAIL: websocket client or group missing" + } + text := "WS 群消息 " + suffix + seq, topic, err := alice.pubWithHeadAndPeer(bob, a.groupTopic, map[string]any{"mime": "text/plain"}, text) + if err != nil { + return "FAIL: " + err.Error() + } + return fmt.Sprintf("seq=%d topic=%s", seq, topic) + }) + a.step("WS-23", "get 群 desc/sub/data", "WS", a.groupTopic, func() string { + if alice == nil || a.groupTopic == "" { + return "FAIL: websocket client or group missing" + } + if _, err := alice.get(a.groupTopic, "desc sub data", map[string]any{"data": map[string]any{"since": 1, "limit": 20}}); err != nil { + return "FAIL: " + err.Error() + } + return "get group ok" + }) + a.step("WS-24", "文件上传 /v0/file/u", "HTTP", "/v0/file/u/", func() string { + fileURL, err := a.uploadFile() + if err != nil { + return "FAIL: " + err.Error() + } + a.fileURL = fileURL + return "url=" + mask(fileURL) + }) + a.step("WS-25", "WS 文件消息下发", "WS", a.p2pTopic, func() string { + if alice == nil || bob == nil || a.fileURL == "" { + return "FAIL: websocket client or file missing" + } + content := map[string]any{ + "txt": "文件", + "ent": []map[string]any{{ + "tp": "EX", + "data": map[string]any{ + "mime": "text/plain", + "name": "imwsaudit.txt", + "ref": a.fileURL, + "size": 24, + }, + }}, + } + seq, topic, err := alice.pubContentWithPeer(bob, a.p2pTopic, map[string]any{"mime": "text/x-drafty", "x-im-schema": 1, "x-im-type": "file"}, content) + if err != nil { + return "FAIL: " + err.Error() + } + return fmt.Sprintf("seq=%d bob_topic=%s", seq, topic) + }) + a.step("WS-26", "WebRTC 通话 invite pub", "WS", a.p2pTopic, func() string { + if alice == nil { + return "FAIL: websocket client missing" + } + callID := "call-" + suffix + seq, err := alice.pubCtrl(a.p2pTopic, map[string]any{"mime": "application/json", "webrtc": "started", "x-im-type": "call"}, map[string]any{"callId": callID, "callType": "audio", "audio": true, "video": false}) + if err != nil { + return "FAIL: " + err.Error() + } + return fmt.Sprintf("seq=%d call_id=%s", seq, callID) + }) + a.step("WS-27", "未 hi 直接 login 被拒绝", "WS", "/v0/channels", func() string { + c, err := a.openWS("no-hi") + if err != nil { + return "FAIL: " + err.Error() + } + defer c.conn.Close() + code, text, err := c.sendCtrl("login-no-hi", map[string]any{"login": map[string]any{"id": "login-no-hi", "scheme": "token", "secret": a.tokenA}}, 5*time.Second) + if err != nil { + return "FAIL: " + err.Error() + } + if code >= 400 { + return fmt.Sprintf("code=%d text=%s", code, text) + } + return fmt.Sprintf("FAIL: unexpected code=%d text=%s", code, text) + }) + a.step("WS-28", "hi 后切换 tenant 被拒绝", "WS", "/v0/channels", func() string { + if alice == nil { + return "FAIL: websocket client missing" + } + code, text, err := alice.sendCtrl("hi-switch", map[string]any{"hi": map[string]any{"id": "hi-switch", "ver": "0.25", "tenant": "other"}}, 5*time.Second) + if err != nil { + return "FAIL: " + err.Error() + } + if code >= 400 { + return fmt.Sprintf("code=%d text=%s", code, text) + } + return fmt.Sprintf("FAIL: unexpected code=%d text=%s", code, text) + }) + a.step("WS-29", "非法 API Key 拒绝建连", "WS", "/v0/channels", func() string { + u := a.wsEndpoint("bad-api-key") + _, resp, err := websocket.DefaultDialer.Dial(u, nil) + if err == nil { + return "FAIL: unexpected websocket upgrade" + } + if resp != nil && resp.StatusCode == http.StatusForbidden { + return "http_status=403" + } + if resp != nil { + return fmt.Sprintf("FAIL: http_status=%d err=%v", resp.StatusCode, err) + } + return "FAIL: " + err.Error() + }) +} + +func (a *audit) step(id, name, method, target string, fn func() string) { + note := fn() + pass := !strings.HasPrefix(note, "FAIL:") + if !pass { + note = strings.TrimSpace(strings.TrimPrefix(note, "FAIL:")) + } + a.results = append(a.results, result{ID: id, Name: name, Method: method, Target: target, Pass: pass, Note: note}) +} + +func (a *audit) register(username, phone, password string) (string, string, error) { + challenge, err := a.verificationCode(phone) + if err != nil { + return "", "", err + } + verification, err := a.verifyCode(challenge) + if err != nil { + return "", "", err + } + c, err := a.openWS("reg-" + username) + if err != nil { + return "", "", err + } + defer c.conn.Close() + if err = c.hi(a.tenantCode); err != nil { + return "", "", err + } + id := "acc-" + username + secret := base64.StdEncoding.EncodeToString([]byte(username + ":" + password)) + if err = c.write(map[string]any{"acc": map[string]any{ + "id": id, + "user": "new", + "scheme": "basic", + "username": username, + "phone": phone, + "secret": secret, + "login": true, + "cred": []map[string]any{{ + "meth": "tel", + "val": phone, + "resp": verification, + "params": map[string]any{"challenge_id": challenge}, + }}, + "desc": map[string]any{"public": map[string]any{"fn": username}}, + }}); err != nil { + return "", "", err + } + ctrl, err := c.waitCtrl(id, 15*time.Second) + if err != nil { + return "", "", err + } + params, _ := ctrl["params"].(map[string]any) + user, token := asString(params["user"]), asString(params["token"]) + if user == "" || token == "" { + return "", "", fmt.Errorf("acc missing user or token") + } + return user, token, nil +} + +func (a *audit) verificationCode(phone string) (string, error) { + status, r, err := a.call("POST", "/v1/im/auth/verification-codes", map[string]any{"scene": "register", "phone": phone}, "") + if err != nil { + return "", err + } + if status != 200 { + return "", fmt.Errorf("verification status=%d reason=%s", status, r.Reason) + } + challenge := asString(r.Data["challenge_id"]) + if challenge == "" { + return "", fmt.Errorf("verification missing challenge_id") + } + return challenge, nil +} + +func (a *audit) verifyCode(challenge string) (string, error) { + status, r, err := a.call("POST", "/v1/im/auth/verification-codes/"+url.PathEscape(challenge)+"/verify", map[string]any{"code": "123456"}, "") + if err != nil { + return "", err + } + if status != 200 { + return "", fmt.Errorf("verify status=%d reason=%s", status, r.Reason) + } + token := asString(r.Data["verification_token"]) + if token == "" { + return "", fmt.Errorf("verify missing verification_token") + } + return token, nil +} + +func (a *audit) createFriendship() (string, error) { + status, r, err := a.call("POST", "/v1/im/friend-requests", map[string]any{"target_user_id": a.userB, "message": "WS 联调好友申请", "source": "search"}, a.tokenA) + if err != nil { + return "", err + } + if status != 200 { + return "", fmt.Errorf("friend create status=%d reason=%s", status, r.Reason) + } + reqID := asString(r.Data["request_id"]) + if reqID == "" { + return "", fmt.Errorf("friend create missing request_id") + } + status, r, err = a.call("POST", "/v1/im/friend-requests/"+url.PathEscape(reqID)+"/accept", map[string]any{"version": 1}, a.tokenB) + if err != nil { + return "", err + } + if status != 200 && status != 409 { + return "", fmt.Errorf("friend accept status=%d reason=%s", status, r.Reason) + } + a.p2pTopic = asString(r.Data["topic"]) + if a.p2pTopic == "" { + return "", fmt.Errorf("friend accept missing topic") + } + return reqID, nil +} + +func (a *audit) createGroup(suffix string) (string, error) { + status, r, err := a.call("POST", "/v1/im/groups", map[string]any{ + "name": "WS协议群" + suffix, + "description": "WS audit group", + "member_user_ids": []string{a.userB}, + "avatar_file_id": "", + }, a.tokenA) + if err != nil { + return "", err + } + if status != 200 { + return "", fmt.Errorf("group create status=%d reason=%s", status, r.Reason) + } + topic := asString(r.Data["topic"]) + if topic == "" { + return "", fmt.Errorf("group create missing topic") + } + return topic, nil +} + +func (a *audit) uploadFile() (string, error) { + var body bytes.Buffer + writer := multipart.NewWriter(&body) + _ = writer.WriteField("id", "upload-"+fmt.Sprintf("%d", time.Now().UnixNano())) + _ = writer.WriteField("auth", "token") + _ = writer.WriteField("secret", a.tokenA) + part, err := writer.CreateFormFile("file", "imwsaudit.txt") + if err != nil { + return "", err + } + if _, err = part.Write([]byte("hello from imwsaudit\n")); err != nil { + return "", err + } + if err = writer.Close(); err != nil { + return "", err + } + req, err := http.NewRequest(http.MethodPost, a.baseURL+"/v0/file/u/", &body) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", writer.FormDataContentType()) + req.Header.Set("X-Tinode-APIKey", a.apiKey) + req.Header.Set("X-Tenant-Code", a.tenantCode) + resp, err := a.client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + var parsed map[string]any + _ = json.Unmarshal(raw, &parsed) + ctrl, _ := parsed["ctrl"].(map[string]any) + if resp.StatusCode != http.StatusOK || asInt(ctrl["code"]) < 200 || asInt(ctrl["code"]) >= 300 { + return "", fmt.Errorf("upload status=%d code=%d text=%s body=%s", resp.StatusCode, asInt(ctrl["code"]), asString(ctrl["text"]), compact(string(raw))) + } + params, _ := ctrl["params"].(map[string]any) + fileURL := asString(params["url"]) + if fileURL == "" { + return "", fmt.Errorf("upload missing url") + } + return fileURL, nil +} + +func (a *audit) openLoggedIn(name, user, token string) (*wsClient, error) { + c, err := a.openWS(name) + if err != nil { + return nil, err + } + if err = c.hi(a.tenantCode); err != nil { + _ = c.conn.Close() + return nil, err + } + loginID := "login-" + strings.ToLower(name) + if err = c.write(map[string]any{"login": map[string]any{ + "id": loginID, + "scheme": "token", + "secret": token, + }}); err != nil { + _ = c.conn.Close() + return nil, err + } + if _, err = c.waitCtrl(loginID, 15*time.Second); err != nil { + _ = c.conn.Close() + return nil, fmt.Errorf("login: %w", err) + } + c.user, c.token = user, token + if err = c.sub("me"); err != nil { + _ = c.conn.Close() + return nil, fmt.Errorf("sub me: %w", err) + } + return c, nil +} + +func (a *audit) openBasicLoggedIn(name, username, password string) (*wsClient, error) { + c, err := a.openWS(name) + if err != nil { + return nil, err + } + if err = c.hi(a.tenantCode); err != nil { + _ = c.conn.Close() + return nil, err + } + loginID := "login-" + strings.ToLower(name) + secret := base64.StdEncoding.EncodeToString([]byte(username + ":" + password)) + if err = c.write(map[string]any{"login": map[string]any{ + "id": loginID, + "scheme": "basic", + "secret": secret, + }}); err != nil { + _ = c.conn.Close() + return nil, err + } + if _, err = c.waitCtrl(loginID, 15*time.Second); err != nil { + _ = c.conn.Close() + return nil, fmt.Errorf("basic login: %w", err) + } + if err = c.sub("me"); err != nil { + _ = c.conn.Close() + return nil, fmt.Errorf("sub me: %w", err) + } + return c, nil +} + +func (a *audit) openWS(name string) (*wsClient, error) { + conn, _, err := websocket.DefaultDialer.Dial(a.wsEndpoint(a.apiKey), nil) + if err != nil { + return nil, err + } + _ = conn.SetReadDeadline(time.Now().Add(30 * time.Second)) + _ = conn.SetWriteDeadline(time.Now().Add(30 * time.Second)) + return &wsClient{name: name, conn: conn}, nil +} + +func (a *audit) wsEndpoint(apiKey string) string { + u, _ := url.Parse(a.wsURL) + q := u.Query() + q.Set("apikey", apiKey) + u.RawQuery = q.Encode() + return u.String() +} + +func (c *wsClient) hi(tenant string) error { + id := "hi-" + strings.ToLower(c.name) + if err := c.write(map[string]any{"hi": map[string]any{"id": id, "ver": "0.25", "ua": "imwsaudit/1.0", "platf": "go", "lang": "zh-CN", "tenant": tenant}}); err != nil { + return err + } + _, err := c.waitCtrl(id, 15*time.Second) + return err +} + +func (c *wsClient) sub(topic string) error { + id := "sub-" + strings.ToLower(c.name) + "-" + strings.ReplaceAll(topic, ":", "_") + if err := c.write(map[string]any{"sub": map[string]any{ + "id": id, + "topic": topic, + "get": map[string]any{"what": "desc sub data", "data": map[string]any{"limit": 10}}, + }}); err != nil { + return err + } + _, err := c.waitCtrl(id, 15*time.Second) + return err +} + +func (c *wsClient) get(topic, what string, extra map[string]any) (map[string]any, error) { + id := "get-" + strings.ToLower(c.name) + "-" + fmt.Sprintf("%d", time.Now().UnixNano()) + get := map[string]any{"id": id, "topic": topic, "what": what} + for k, v := range extra { + get[k] = v + } + if err := c.write(map[string]any{"get": get}); err != nil { + return nil, err + } + if _, err := c.waitCtrl(id, 15*time.Second); err != nil { + return nil, err + } + meta, err := c.wait(func(msg map[string]any) bool { + m, ok := msg["meta"].(map[string]any) + return ok && asString(m["topic"]) == topic + }, 5*time.Second) + if err != nil { + return nil, err + } + return meta["meta"].(map[string]any), nil +} + +func (c *wsClient) setPrivate(topic string, private map[string]any) error { + id := "set-" + strings.ToLower(c.name) + "-" + fmt.Sprintf("%d", time.Now().UnixNano()) + if err := c.write(map[string]any{"set": map[string]any{ + "id": id, + "topic": topic, + "sub": map[string]any{"private": private}, + }}); err != nil { + return err + } + ctrl, err := c.waitCtrlRaw(id, 15*time.Second) + if err != nil { + return err + } + code := asInt(ctrl["code"]) + if (code >= 200 && code < 300) || code == 304 { + return nil + } + return fmt.Errorf("ctrl id=%s code=%d text=%s", id, code, asString(ctrl["text"])) +} + +func (c *wsClient) pubAndPeerReceives(peer *wsClient, topic, text string) (int, string, error) { + id := "pub-" + time.Now().Format("150405.000000000") + if err := c.write(map[string]any{"pub": map[string]any{ + "id": id, + "topic": topic, + "head": map[string]any{"mime": "text/plain"}, + // Keep this as native Tinode text content. Product-specific x-im + // headers should be covered by a separate WS message-header matrix. + "content": text, + }}); err != nil { + return 0, "", err + } + if _, err := c.waitCtrl(id, 15*time.Second); err != nil { + return 0, "", fmt.Errorf("pub ctrl: %w", err) + } + msg, err := peer.wait(func(msg map[string]any) bool { + d, ok := msg["data"].(map[string]any) + return ok && asString(d["content"]) == text + }, 15*time.Second) + if err != nil { + return 0, "", fmt.Errorf("peer data: %w", err) + } + d := msg["data"].(map[string]any) + seq := asInt(d["seq"]) + if seq <= 0 { + return 0, "", fmt.Errorf("data missing seq") + } + return seq, asString(d["topic"]), nil +} + +func (c *wsClient) pubWithHeadAndPeer(peer *wsClient, topic string, head map[string]any, text string) (int, string, error) { + return c.pubContentWithPeer(peer, topic, head, text) +} + +func (c *wsClient) pubContentWithPeer(peer *wsClient, topic string, head map[string]any, content any) (int, string, error) { + seq, err := c.pubCtrl(topic, head, content) + if err != nil { + return 0, "", err + } + msg, err := peer.wait(func(msg map[string]any) bool { + d, ok := msg["data"].(map[string]any) + if !ok || asInt(d["seq"]) != seq { + return false + } + if s, ok := content.(string); ok { + return asString(d["content"]) == s + } + return true + }, 15*time.Second) + if err != nil { + return 0, "", fmt.Errorf("peer data: %w", err) + } + d := msg["data"].(map[string]any) + return seq, asString(d["topic"]), nil +} + +func (c *wsClient) pubCtrl(topic string, head map[string]any, content any) (int, error) { + id := "pub-" + time.Now().Format("150405.000000000") + ctrl, err := c.writeAndWaitCtrl(id, map[string]any{"pub": map[string]any{ + "id": id, + "topic": topic, + "head": head, + "content": content, + }}, 15*time.Second) + if err != nil { + return 0, fmt.Errorf("pub ctrl: %w", err) + } + params, _ := ctrl["params"].(map[string]any) + seq := asInt(params["seq"]) + return seq, nil +} + +func (c *wsClient) note(topic, what string, seq int) error { + return c.write(map[string]any{"note": map[string]any{"topic": topic, "what": what, "seq": seq}}) +} + +func (c *wsClient) delMsg(topic string, seq int) error { + id := "del-" + strings.ToLower(c.name) + "-" + fmt.Sprintf("%d", time.Now().UnixNano()) + return c.writeAndWaitOK(id, map[string]any{"del": map[string]any{ + "id": id, + "topic": topic, + "what": "msg", + "delseq": []map[string]any{{"low": seq, "hi": seq + 1}}, + }}, 15*time.Second) +} + +func (c *wsClient) leave(topic string, unsub bool) error { + id := "leave-" + strings.ToLower(c.name) + "-" + fmt.Sprintf("%d", time.Now().UnixNano()) + return c.writeAndWaitOK(id, map[string]any{"leave": map[string]any{ + "id": id, + "topic": topic, + "unsub": unsub, + }}, 15*time.Second) +} + +func (c *wsClient) write(msg map[string]any) error { + _ = c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + return c.conn.WriteJSON(msg) +} + +func (c *wsClient) sendCtrl(id string, msg map[string]any, timeout time.Duration) (int, string, error) { + if err := c.write(msg); err != nil { + return 0, "", err + } + ctrl, err := c.waitCtrlRaw(id, timeout) + if err != nil { + return 0, "", err + } + return asInt(ctrl["code"]), asString(ctrl["text"]), nil +} + +func (c *wsClient) writeAndWaitOK(id string, msg map[string]any, timeout time.Duration) error { + _, err := c.writeAndWaitCtrl(id, msg, timeout) + return err +} + +func (c *wsClient) writeAndWaitCtrl(id string, msg map[string]any, timeout time.Duration) (map[string]any, error) { + if err := c.write(msg); err != nil { + return nil, err + } + return c.waitCtrl(id, timeout) +} + +func (c *wsClient) waitCtrl(id string, timeout time.Duration) (map[string]any, error) { + ctrl, err := c.waitCtrlRaw(id, timeout) + if err != nil { + return nil, err + } + code := asInt(ctrl["code"]) + if code < 200 || code >= 300 { + return ctrl, fmt.Errorf("ctrl id=%s code=%d text=%s", id, code, asString(ctrl["text"])) + } + return ctrl, nil +} + +func (c *wsClient) waitCtrlRaw(id string, timeout time.Duration) (map[string]any, error) { + msg, err := c.wait(func(msg map[string]any) bool { + ctrl, ok := msg["ctrl"].(map[string]any) + return ok && asString(ctrl["id"]) == id + }, timeout) + if err != nil { + return nil, err + } + return msg["ctrl"].(map[string]any), nil +} + +func (c *wsClient) wait(match func(map[string]any) bool, timeout time.Duration) (map[string]any, error) { + for i := 0; i < len(c.pending); i++ { + if match(c.pending[i]) { + msg := c.pending[i] + c.pending = append(c.pending[:i], c.pending[i+1:]...) + return msg, nil + } + } + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + _ = c.conn.SetReadDeadline(deadline) + var msg map[string]any + if err := c.conn.ReadJSON(&msg); err != nil { + return nil, err + } + if match(msg) { + return msg, nil + } + c.pending = append(c.pending, msg) + } + return nil, fmt.Errorf("timeout") +} + +func (a *audit) call(method, path string, body any, token string) (int, imResp, error) { + var reader io.Reader + if body != nil { + b, _ := json.Marshal(body) + reader = bytes.NewReader(b) + } + req, err := http.NewRequest(method, a.baseURL+path, reader) + if err != nil { + return 0, imResp{}, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Tinode-APIKey", a.apiKey) + req.Header.Set("X-Tenant-Code", a.tenantCode) + req.Header.Set("X-IM-Schema", "1") + req.Header.Set("X-Request-ID", "req-ws-audit-"+fmt.Sprintf("%d", time.Now().UnixNano())) + if method != http.MethodGet { + req.Header.Set("Idempotency-Key", "idem-ws-audit-"+fmt.Sprintf("%d", time.Now().UnixNano())) + } + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("X-Tinode-Auth", "token "+base64.StdEncoding.EncodeToString([]byte(token))) + } + resp, err := a.client.Do(req) + if err != nil { + return 0, imResp{}, err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + var parsed imResp + _ = json.Unmarshal(raw, &parsed) + return resp.StatusCode, parsed, nil +} + +func (a *audit) report() string { + total, passed := len(a.results), 0 + for _, r := range a.results { + if r.Pass { + passed++ + } + } + sort.SliceStable(a.results, func(i, j int) bool { return a.results[i].ID < a.results[j].ID }) + var b strings.Builder + fmt.Fprintf(&b, "# IM WebSocket 测试环境专项联调结果（2026-08-21）\n\n") + fmt.Fprintf(&b, "## 1. 测试范围\n\n") + fmt.Fprintf(&b, "- 测试时间：`%s`\n", time.Now().Format("2006-01-02 15:04:05 -0700")) + fmt.Fprintf(&b, "- HTTP：`%s`\n", a.baseURL) + fmt.Fprintf(&b, "- WebSocket：`%s`\n", a.wsURL) + fmt.Fprintf(&b, "- 租户：`%s`\n", a.tenantCode) + fmt.Fprintf(&b, "- 用户 A：`%s` / `%s` / `%s`\n", a.userA, a.usernameA, a.phoneA) + fmt.Fprintf(&b, "- 用户 B：`%s` / `%s` / `%s`\n", a.userB, a.usernameB, a.phoneB) + fmt.Fprintf(&b, "- API Key 与 Token 已脱敏，不在文档中展开。\n\n") + fmt.Fprintf(&b, "## 2. 总体结论\n\n") + fmt.Fprintf(&b, "本轮专项验证 Tinode WebSocket 真实协议链路，共执行 `%d` 项，通过 `%d` 项，未通过 `%d` 项。\n\n", total, passed, total-passed) + fmt.Fprintf(&b, "| ID | 场景 | 方法 | 目标 | 结果 | 备注 |\n| --- | --- | --- | --- | --- | --- |\n") + for _, r := range a.results { + status := "通过" + if !r.Pass { + status = "未通过" + } + fmt.Fprintf(&b, "| `%s` | %s | `%s` | `%s` | %s | %s |\n", r.ID, esc(r.Name), r.Method, esc(r.Target), status, esc(r.Note)) + } + fmt.Fprintf(&b, "\n## 3. 覆盖说明\n\n") + fmt.Fprintf(&b, "- 已覆盖：API Key Upgrade、`hi(tenant)`、`acc`、`basic login`、`token login`、重复登录拒绝、`sub me/fnd/p2p/grp`、`get desc/sub/data/del/tags/cred`、`set sub.private`、P2P/群 `pub`、产品 `x-im-*` 基础头、`note recv/read/kp`、`del msg`、`leave`、文件上传入口、文件消息、WebRTC invite、未 hi 顺序错误、切换租户拒绝、非法 API Key 拒绝。\n") + fmt.Fprintf(&b, "- 仍需前端/真机补充：断线重连恢复、Long Polling 长轮询持续收包、真实音视频媒体协商质量、离线 Push 到达与点击恢复。\n") + fmt.Fprintf(&b, "\n## 4. 测试产生的数据\n\n") + fmt.Fprintf(&b, "- P2P Topic：`%s`\n", a.p2pTopic) + fmt.Fprintf(&b, "- Bob 收到的 Topic：`%s`\n", a.bobTopic) + fmt.Fprintf(&b, "- 消息 Seq：`%s`\n", a.messageID) + return b.String() +} + +func asString(v any) string { + switch t := v.(type) { + case string: + return t + default: + return "" + } +} + +func asInt(v any) int { + switch t := v.(type) { + case int: + return t + case float64: + return int(t) + case json.Number: + n, _ := t.Int64() + return int(n) + default: + return 0 + } +} + +func mask(s string) string { + if len(s) <= 8 { + return s + } + return s[:4] + "..." + s[len(s)-4:] +} + +func esc(s string) string { + s = strings.ReplaceAll(s, "|", "\\|") + s = strings.ReplaceAll(s, "\n", " ") + return s +} + +func compact(s string) string { + s = strings.TrimSpace(strings.ReplaceAll(s, "\n", " ")) + if len(s) > 220 { + return s[:220] + "..." + } + return s +} diff --git a/docs/test/im_websocket_testenv_results_2026-08-21.md b/docs/test/im_websocket_testenv_results_2026-08-21.md new file mode 100644 index 0000000..bd80fa9 --- /dev/null +++ b/docs/test/im_websocket_testenv_results_2026-08-21.md @@ -0,0 +1,58 @@ +# IM WebSocket 测试环境专项联调结果（2026-08-21） + +## 1. 测试范围 + +- 测试时间：`2026-08-21 00:39:01 +0800` +- HTTP：`https://api.118im.com` +- WebSocket：`wss://api.118im.com/v0/channels` +- 租户：`default` +- 用户 A：`usr7-zA3UMrw7c` / `wsa459000` / `15794459000` +- 用户 B：`usrnDXQ3A2oQP0` / `wsb459000` / `15894459000` +- API Key 与 Token 已脱敏，不在文档中展开。 + +## 2. 总体结论 + +本轮专项验证 Tinode WebSocket 真实协议链路，共执行 `29` 项，通过 `27` 项，未通过 `2` 项。 + +| ID | 场景 | 方法 | 目标 | 结果 | 备注 | +| --- | --- | --- | --- | --- | --- | +| `WS-01` | 注册用户 A: HTTP 验证码 + WS acc | `WS` | `/v0/channels` | 通过 | user=usr7-zA3UMrw7c token=AwEA...18Om | +| `WS-02` | 注册用户 B: HTTP 验证码 + WS acc | `WS` | `/v0/channels` | 通过 | user=usrnDXQ3A2oQP0 token=AwEA...NX6b | +| `WS-03` | HTTP 建立好友关系生成 P2P Topic | `HTTP` | `/v1/im/friend-requests` | 通过 | request_id=frq_KAOkEfarCOA topic=p2p7-zA3UMrw7ecNdDcDahA_Q | +| `WS-04` | A token login + sub me | `WS` | `/v0/channels` | 通过 | login user=usr7-zA3UMrw7c | +| `WS-05` | B token login + sub me | `WS` | `/v0/channels` | 通过 | login user=usrnDXQ3A2oQP0 | +| `WS-06` | A/B 订阅 P2P Topic | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | topic=p2p7-zA3UMrw7ecNdDcDahA_Q | +| `WS-07` | A pub 文本消息，B 收到 data | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | seq=1 bob_topic=usr7-zA3UMrw7c | +| `WS-08` | B 发送 recv/read 回执 | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | what=read seq=1 | +| `WS-09` | basic login + sub me | `WS` | `/v0/channels` | 通过 | basic login ok | +| `WS-10` | 重复 login 被拒绝 | `WS` | `/v0/channels` | 通过 | code=409 text=already authenticated | +| `WS-11` | 订阅 fnd 搜索 Topic | `WS` | `fnd` | 通过 | sub fnd ok | +| `WS-12` | get me desc/sub/tags/cred | `WS` | `me` | 通过 | get me ok | +| `WS-13` | get P2P desc/sub/data/del | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | get p2p ok | +| `WS-14` | set P2P private 设置 | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | set private ok | +| `WS-15` | pub 产品 x-im 基础头 | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | seq=2 bob_topic=usr7-zA3UMrw7c | +| `WS-16` | pub 产品 x-im-client-mid 幂等头 | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 未通过 | pub ctrl: ctrl id=pub-003857.244320000 code=500 text=internal error | +| `WS-17` | note kp 输入中通知 | `WS` | `usr7-zA3UMrw7c` | 通过 | what=kp | +| `WS-18` | del 当前用户消息视图 | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | deleted seq=1 | +| `WS-19` | leave P2P detach 后重新订阅 | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | leave/resub ok | +| `WS-20` | HTTP 创建群供 WS 群协议测试 | `HTTP` | `/v1/im/groups` | 通过 | topic=grpURzKYanXB1Y | +| `WS-21` | A/B 订阅群 Topic | `WS` | `grpURzKYanXB1Y` | 通过 | sub group ok | +| `WS-22` | 群 pub 文本广播 | `WS` | `grpURzKYanXB1Y` | 通过 | seq=1 topic=grpURzKYanXB1Y | +| `WS-23` | get 群 desc/sub/data | `WS` | `grpURzKYanXB1Y` | 通过 | get group ok | +| `WS-24` | 文件上传 /v0/file/u | `HTTP` | `/v0/file/u/` | 通过 | url=/v0/....bin | +| `WS-25` | WS 文件消息下发 | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 通过 | seq=3 bob_topic=usr7-zA3UMrw7c | +| `WS-26` | WebRTC 通话 invite pub | `WS` | `p2p7-zA3UMrw7ecNdDcDahA_Q` | 未通过 | pub ctrl: ctrl id=pub-003859.867810000 code=501 text=not implemented | +| `WS-27` | 未 hi 直接 login 被拒绝 | `WS` | `/v0/channels` | 通过 | code=409 text=command out of sequence | +| `WS-28` | hi 后切换 tenant 被拒绝 | `WS` | `/v0/channels` | 通过 | code=409 text=command out of sequence | +| `WS-29` | 非法 API Key 拒绝建连 | `WS` | `/v0/channels` | 通过 | http_status=403 | + +## 3. 覆盖说明 + +- 已覆盖：API Key Upgrade、`hi(tenant)`、`acc`、`basic login`、`token login`、重复登录拒绝、`sub me/fnd/p2p/grp`、`get desc/sub/data/del/tags/cred`、`set sub.private`、P2P/群 `pub`、产品 `x-im-*` 基础头、`note recv/read/kp`、`del msg`、`leave`、文件上传入口、文件消息、WebRTC invite、未 hi 顺序错误、切换租户拒绝、非法 API Key 拒绝。 +- 仍需前端/真机补充：断线重连恢复、Long Polling 长轮询持续收包、真实音视频媒体协商质量、离线 Push 到达与点击恢复。 + +## 4. 测试产生的数据 + +- P2P Topic：`p2p7-zA3UMrw7ecNdDcDahA_Q` +- Bob 收到的 Topic：`usr7-zA3UMrw7c` +- 消息 Seq：`1` diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 7228a0c..17206ba 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -3568,10 +3568,11 @@ func (a *adapter) MessageSave(msg *t.Message) error { } // store assignes message ID, but we don't use it. Message IDs are not used anywhere. // Using a sequential ID provided by the database. + clientMsgID := sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""} res, err := a.db.ExecContext(ctx, - "INSERT INTO im_messages(createdAt,updatedAt,seqid,topic,`from`,head,content) VALUES(?,?,?,?,?,?,?)", + "INSERT INTO im_messages(createdAt,updatedAt,seqid,topic,`from`,client_msg_id,version,head,content) VALUES(?,?,?,?,?,?,?,?,?)", msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, - store.DecodeUid(t.ParseUid(msg.From)), msg.Head, common.ToJSON(msg.Content)) + store.DecodeUid(t.ParseUid(msg.From)), clientMsgID, messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content)) if err == nil { id, _ := res.LastInsertId() // Replacing ID given by store by ID given by the DB. diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 9bc0b04..7977299 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -1622,7 +1622,7 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error { return t.ErrMalformed } if tenantAdp != a { - return tenantAdp.MessageSave(msg) + return tenantAdp.TenantMessageSave(id, msg) } if a.tenantScopedDB { return a.MessageSave(msg) diff --git a/server/tinode.conf b/server/tinode.conf index 8ba5d18..4d0455b 100644 --- a/server/tinode.conf +++ b/server/tinode.conf @@ -36,6 +36,16 @@ } }, + "webrtc": { + "enabled": true, + "call_establishment_timeout": 30, + "ice_servers": [ + { + "urls": ["stun:stun.l.google.com:19302"] + } + ] + }, + "tls": { "enabled": false, "http_redirect": "",