package main import ( "bytes" "crypto/hmac" "crypto/sha256" "encoding/base64" "encoding/hex" "encoding/json" "errors" "fmt" "io" "net/http" "net/url" "runtime/debug" "strconv" "strings" "sync" "time" "github.com/tinode/chat/server/auth" "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" ) const ( internalCapabilityAdminQuery = "admin_query" internalCapabilityCultivate = "cultivate" internalHeaderService = "X-Service-Name" internalHeaderTimestamp = "X-Timestamp" internalHeaderNonce = "X-Nonce" internalHeaderSignature = "X-Signature" ) type internalAPIConfig struct { Enabled bool `json:"enabled"` TimeWindowSeconds int `json:"time_window_seconds"` Services map[string]internalAPIServiceConfig `json:"services"` } type internalAPIServiceConfig struct { Secret string `json:"secret"` Allow []string `json:"allow"` } type internalAPIRuntime struct { config internalAPIConfig nonces sync.Map } var internalAPI internalAPIRuntime func initInternalAPI(raw json.RawMessage) error { internalAPI = internalAPIRuntime{} if len(raw) == 0 || string(raw) == "null" { return nil } var cfg internalAPIConfig if err := json.Unmarshal(raw, &cfg); err != nil { return err } if cfg.TimeWindowSeconds <= 0 { cfg.TimeWindowSeconds = 300 } if cfg.Services == nil { cfg.Services = map[string]internalAPIServiceConfig{} } internalAPI.config = cfg if cfg.Enabled { logs.Info.Printf("Internal API enabled for %d service(s)", len(cfg.Services)) } return nil } func serveInternalAdminQuery(wrt http.ResponseWriter, req *http.Request, prefix string) { defer func() { if recovered := recover(); recovered != nil { logs.Err.Printf("internal admin query panic path:%s err:%v stack:%s", req.URL.RequestURI(), recovered, debug.Stack()) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) } }() if req.Method != http.MethodGet && req.Method != http.MethodPost { writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) return } if !authorizeInternalAPI(wrt, req, internalCapabilityAdminQuery) { return } tail := strings.Trim(strings.TrimPrefix(req.URL.Path, prefix), "/") parts := splitInternalPath(tail) if len(parts) < 3 || parts[0] != "tenants" { writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil) return } tenant, ok := resolveInternalTenant(wrt, req, parts[1]) if !ok { return } if req.Method == http.MethodPost && !isInternalAdminPostRoute(parts) { writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) return } switch { case len(parts) == 4 && parts[2] == "dashboard" && parts[3] == "overview": serveInternalDashboardOverview(wrt, req, tenant.ID) case len(parts) == 4 && parts[2] == "dashboard" && parts[3] == "trends": serveInternalDashboardTrends(wrt, req, tenant.ID) case len(parts) == 3 && parts[2] == "private-messages": serveInternalAdminResourceList(wrt, req, tenant.ID, "private-messages", "") case len(parts) == 3 && parts[2] == "group-messages": serveInternalAdminResourceList(wrt, req, tenant.ID, "group-messages", "") case req.Method == http.MethodPost && len(parts) == 3 && parts[2] == "users": serveInternalUserCreate(wrt, req, tenant.ID) case req.Method == http.MethodPost && len(parts) == 5 && parts[2] == "users" && parts[4] == "action": serveInternalUserAction(wrt, req, tenant.ID, parts[3]) case req.Method == http.MethodPost && len(parts) == 4 && parts[2] == "login-lockouts" && parts[3] == "release": serveInternalLoginLockoutRelease(wrt, req, tenant.ID) case req.Method == http.MethodGet && len(parts) == 3 && parts[2] == "users": serveInternalAdminResourceList(wrt, req, tenant.ID, "users", "") case len(parts) == 4 && parts[2] == "users": serveInternalUserQuery(wrt, req, tenant.ID, parts[3]) case len(parts) == 5 && parts[2] == "users" && parts[4] == "devices": serveInternalAdminResourceList(wrt, req, tenant.ID, "user_devices", parts[3]) case len(parts) == 3 && parts[2] == "groups": serveInternalAdminResourceList(wrt, req, tenant.ID, "groups", "") case len(parts) == 4 && parts[2] == "groups": serveInternalGroupQuery(wrt, req, tenant.ID, parts[3]) case len(parts) == 5 && parts[2] == "groups" && parts[4] == "members": serveInternalGroupMembersQuery(wrt, req, tenant.ID, parts[3]) case len(parts) == 5 && parts[2] == "groups" && internalGroupResourceSpecs[parts[4]] != nil: serveInternalAdminResourceList(wrt, req, tenant.ID, "group_"+parts[4], parts[3]) case len(parts) == 5 && parts[2] == "messages": serveInternalMessageQuery(wrt, req, tenant.ID, parts[3], parts[4]) case len(parts) == 3 && parts[2] == "files": serveInternalAdminResourceList(wrt, req, tenant.ID, "files", "") case len(parts) == 4 && parts[2] == "files": serveInternalFileQuery(wrt, req, tenant.ID, parts[3]) case len(parts) == 5 && parts[2] == "files" && parts[4] == "links": serveInternalAdminResourceList(wrt, req, tenant.ID, "file_links", parts[3]) case len(parts) == 3 && internalTenantResourceSpecs[parts[2]] != nil: serveInternalAdminResourceList(wrt, req, tenant.ID, parts[2], "") case len(parts) == 4 && internalTenantResourceSpecs[parts[2]] != nil: serveInternalAdminResourceDetail(wrt, req, tenant.ID, parts[2], parts[3]) case len(parts) == 4 && parts[2] == "reports" && parts[3] == "abuse": writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil) case len(parts) == 6 && parts[2] == "reports" && parts[3] == "abuse" && parts[5] == "target": serveInternalAbuseTargetQuery(wrt, req, tenant.ID, parts[4]) case len(parts) == 4 && parts[2] == "outbox" && internalOutboxResourceSpecs[parts[3]] != nil: serveInternalAdminResourceList(wrt, req, tenant.ID, "outbox_"+parts[3], "") case len(parts) == 5 && parts[2] == "outbox" && internalOutboxResourceSpecs[parts[3]] != nil: serveInternalAdminResourceDetail(wrt, req, tenant.ID, "outbox_"+parts[3], parts[4]) case len(parts) == 4 && parts[2] == "dashboard" && parts[3] == "summary": serveInternalDashboardSummary(wrt, req, tenant.ID) case len(parts) == 4 && parts[2] == "kafka" && parts[3] == "dead-letters": serveInternalKafkaDeadLettersQuery(wrt, req, tenant.ID) case len(parts) == 5 && parts[2] == "kafka" && parts[3] == "dead-letters": serveInternalKafkaDeadLetterQuery(wrt, req, tenant.ID, parts[4]) default: writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil) } } func isInternalAdminPostRoute(parts []string) bool { return (len(parts) == 3 && parts[2] == "users") || (len(parts) == 5 && parts[2] == "users" && parts[4] == "action") || (len(parts) == 4 && parts[2] == "login-lockouts" && parts[3] == "release") } type internalUserCreateRequest struct { Username string `json:"username"` Nickname string `json:"nickname"` Phone string `json:"phone"` Password string `json:"password"` } type internalUserActionRequest struct { EventType string `json:"event_type"` ActorUserID string `json:"actor_user_id"` DeviceID string `json:"device_id"` PasswordHash string `json:"password_hash"` Reason string `json:"reason"` Hard bool `json:"hard"` } func serveInternalUserCreate(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) { var body internalUserCreateRequest if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&body); err != nil { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) return } body.Username = strings.ToLower(strings.TrimSpace(body.Username)) body.Nickname = strings.TrimSpace(body.Nickname) body.Phone = strings.TrimSpace(body.Phone) if body.Username == "" { var err error body.Username, err = randomRegistrationUsername() if err != nil { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } } if err := validateRegistrationUsername(body.Username); err != nil || len([]rune(body.Password)) < 3 || len([]byte(body.Password)) > 72 { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "用户名或密码格式不正确", nil) return } if body.Phone != "" { phone, err := normalizeIMPhone(body.Phone, "CN") if err != nil { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "手机号格式不正确", nil) return } body.Phone = phone } authHandler := store.Store.GetLogicalAuthHandler("basic") if authHandler == nil { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "basic authentication unavailable", nil) return } authCtx := auth.AuthContext{TenantID: tenantID, RemoteAddr: req.RemoteAddr} secret := []byte(body.Username + ":" + body.Password) if unique, err := authHandler.IsUnique(authCtx, secret); err != nil || !unique { writeIMError(wrt, req, http.StatusConflict, 30001, "DUPLICATE_USER", "用户名已存在", nil) return } user := &types.User{ TenantID: tenantID, State: types.StateOK, Username: body.Username, Phone: body.Phone, Public: map[string]any{"fn": firstNonEmpty(body.Nickname, body.Username)}, } user.Access.Auth = getDefaultAccess(types.TopicCatP2P, true, false) | getDefaultAccess(types.TopicCatGrp, true, false) user.Access.Anon = getDefaultAccess(types.TopicCatP2P, false, false) | getDefaultAccess(types.TopicCatGrp, false, false) created, err := store.Users.Create(tenantID, user, nil) if err != nil { writeInternalQueryError(wrt, req, err) return } if _, err = authHandler.AddRecord(authCtx, &auth.Rec{TenantID: tenantID, Uid: created.Uid()}, secret); err != nil { _ = store.Users.Delete(tenantID, created.Uid(), true) writeInternalQueryError(wrt, req, err) return } displayNo := int64(0) if row, err := store.InternalAdminQueries.QueryOne(tenantID, "SELECT account_no FROM im_users WHERE id=?", []any{store.DecodeUid(created.Uid())}); err == nil { displayNo, _ = numericID(row["account_no"]) } else { logs.Warn.Printf("internal user create account number query failed tenant:%d user:%d err:%v", tenantID, store.DecodeUid(created.Uid()), err) } writeIMSuccess(wrt, req, map[string]any{ "user_id": store.DecodeUid(created.Uid()), "display_no": displayNo, "username": created.Username, "nickname": body.Nickname, "phone": created.Phone, }) } func serveInternalUserAction(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, rawUserID string) { var body internalUserActionRequest if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&body); err != nil { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) return } eventType := strings.TrimSpace(body.EventType) if !isInternalUserActionSupported(eventType) { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "unsupported user action", map[string]any{"event_type": eventType}) return } target := parseKafkaCommandUser(rawUserID) if target.IsZero() { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "用户ID不能为空", nil) return } payload := adminCommandPayload{ CommandID: firstNonEmpty(strings.TrimSpace(req.Header.Get("X-Request-ID")), "internal-"+strconv.FormatInt(time.Now().UTC().UnixNano(), 10)), ActorUserID: strings.TrimSpace(body.ActorUserID), UserID: strconv.FormatInt(internalAdminUIDRaw(target), 10), TargetID: strconv.FormatInt(internalAdminUIDRaw(target), 10), DeviceID: strings.TrimSpace(body.DeviceID), PasswordHash: strings.TrimSpace(body.PasswordHash), Reason: strings.TrimSpace(body.Reason), Hard: body.Hard, } envelope := kafkaEnvelope{ EventID: payload.CommandID, TraceID: strings.TrimSpace(req.Header.Get("X-Request-ID")), EventType: eventType, BizType: "user", BizID: strconv.FormatInt(internalAdminUIDRaw(target), 10), OccurredAt: time.Now().UTC(), } var ( result map[string]any err error ) if eventType == adminCommandUserPasswordReset { result, err = executeKafkaUserPasswordResetCommand(tenantID, payload) } else { result, err = executeKafkaUserGovernanceCommand(tenantID, envelope, payload, parseKafkaCommandUser(payload.ActorUserID)) } if err != nil { writeInternalQueryError(wrt, req, err) return } if err := enqueueKafkaAuditEvent(req.Context(), tenantID, "admin", parseKafkaCommandUser(payload.ActorUserID), eventType, "internal_user_action", payload.CommandID, "succeeded", "", result); err != nil { logs.Warn.Printf("internal user action audit enqueue failed tenant:%d user:%d event:%s err:%v", tenantID, internalAdminUIDRaw(target), eventType, err) } writeIMSuccess(wrt, req, map[string]any{ "user_id": internalAdminUIDRaw(target), "event_type": eventType, "status": "succeeded", "result": result, }) } func isInternalUserActionSupported(eventType string) bool { switch strings.TrimSpace(eventType) { case adminCommandUserBan, adminCommandUserUnban, adminCommandUserFreeze, adminCommandUserUnfreeze, adminCommandUserDeviceKick, adminCommandUserLogoutAll, adminCommandUserCancel, adminCommandUserDelete, adminCommandUserPasswordReset: return true default: return false } } type internalLoginLockoutReleaseRequest struct { AccountHash string `json:"account_hash"` } // serveInternalLoginLockoutRelease clears the login failure lockout for one // account so the user can sign in again immediately. func serveInternalLoginLockoutRelease(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) { var body internalLoginLockoutReleaseRequest if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&body); err != nil { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) return } accountHash := strings.ToLower(strings.TrimSpace(body.AccountHash)) if len(accountHash) != 64 { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "account_hash 格式不正确", nil) return } rec, err := store.LoginLockouts.Get(tenantID, accountHash) if err != nil && !errors.Is(err, types.ErrNotFound) && !errors.Is(err, types.ErrUnsupported) { writeInternalQueryError(wrt, req, err) return } if err = store.LoginLockouts.Release(tenantID, accountHash, time.Now().UTC().Round(time.Millisecond)); err != nil { writeInternalQueryError(wrt, req, err) return } logs.Info.Printf("login lockout released tenant:%d account_hash:%s had_lock:%v", tenantID, accountHash, rec != nil && rec.LockedUntil != nil) writeIMSuccess(wrt, req, map[string]any{"account_hash": accountHash, "status": "released"}) } func authorizeInternalAPI(wrt http.ResponseWriter, req *http.Request, capability string) bool { if !internalAPI.config.Enabled { writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil) return false } serviceName := strings.TrimSpace(req.Header.Get(internalHeaderService)) timestampRaw := strings.TrimSpace(req.Header.Get(internalHeaderTimestamp)) nonce := strings.TrimSpace(req.Header.Get(internalHeaderNonce)) signature := strings.TrimSpace(req.Header.Get(internalHeaderSignature)) service, ok := internalAPI.config.Services[serviceName] if !ok || service.Secret == "" || timestampRaw == "" || nonce == "" || signature == "" { writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal signature required", nil) return false } if !internalServiceAllows(service, capability) { writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) return false } timestamp, err := strconv.ParseInt(timestampRaw, 10, 64) if err != nil { writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "invalid internal timestamp", nil) return false } window := time.Duration(internalAPI.config.TimeWindowSeconds) * time.Second if delta := time.Since(time.Unix(timestamp, 0)); delta > window || delta < -window { writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal signature expired", nil) return false } bodyHash, err := hashInternalRequestBody(req) if err != nil { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return false } signingText := strings.Join([]string{ req.Method, req.URL.RequestURI(), bodyHash, timestampRaw, nonce, }, "\n") if !verifyInternalSignature(service.Secret, signingText, signature) { writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "invalid internal signature", nil) return false } requestClaim := internalRequestClaimValue(signingText) if !claimInternalNonce(serviceName, nonce, requestClaim, time.Now().Add(window)) { writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal nonce replayed", nil) return false } return true } func splitInternalPath(tail string) []string { if tail == "" { return nil } rawParts := strings.Split(tail, "/") parts := make([]string, 0, len(rawParts)) for _, part := range rawParts { if part == "" { continue } decoded, err := url.PathUnescape(part) if err != nil { decoded = part } parts = append(parts, decoded) } return parts } func resolveInternalTenant(wrt http.ResponseWriter, req *http.Request, tenantCode string) (*types.Tenant, bool) { if tenantCode == "" || store.Tenants == nil { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "tenant_code"}) return nil, false } tenant, err := store.Tenants.GetByCode(tenantCode) if err != nil { if errors.Is(err, types.ErrNotFound) { writeIMError(wrt, req, http.StatusNotFound, 30010, imReasonTenantUnavailable, "tenant unavailable", nil) return nil, false } writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return nil, false } if !tenant.IsActive() { writeIMError(wrt, req, http.StatusForbidden, 30010, imReasonTenantUnavailable, "tenant unavailable", nil) return nil, false } return tenant, true } func serveInternalUserQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, userID string) { userDBID, ok := parseInternalAdminDBID(userID) if !ok { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"}) return } if spec, ok := internalAdminQuerySpec("users"); ok { query := spec.Select + " " + spec.From + " WHERE u.id=?" if row, err := store.InternalAdminQueries.QueryOne(tenantID, query, []any{userDBID}); err == nil { writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row)) return } else { logs.Warn.Printf("internal user query sql failed tenant:%d user:%d err:%v", tenantID, userDBID, err) } } uid := store.EncodeUid(userDBID) user, err := store.Users.Get(tenantID, uid) if err != nil { logs.Warn.Printf("internal user query fallback failed tenant:%d user:%d err:%v", tenantID, userDBID, err) writeInternalQueryError(wrt, req, err) return } if user == nil { writeInternalQueryError(wrt, req, types.ErrUserNotFound) return } writeIMSuccess(wrt, req, map[string]any{ "user_id": userDBID, "state": user.State, "state_at": user.StateAt, "username": user.Username, "phone": user.Phone, "public": user.Public, "trusted": user.Trusted, "last_seen": user.LastSeen, "user_agent": user.UserAgent, "tags": user.Tags, "created_at": user.CreatedAt, "updated_at": user.UpdatedAt, }) } func serveInternalGroupQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic string) { if topic == "" || store.GroupBusiness == nil { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"}) return } if spec, ok := internalAdminQuerySpec("groups"); ok { query := spec.Select + " " + spec.From + " WHERE t.name=?" if row, err := store.InternalAdminQueries.QueryOne(tenantID, query, []any{topic}); err == nil { writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row)) return } } settings, err := store.GroupBusiness.GetSettings(tenantID, topic) if err != nil { writeInternalQueryError(wrt, req, err) return } writeIMSuccess(wrt, req, internalGroupSettingsPayload(settings)) } func serveInternalGroupMembersQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic string) { if topic == "" || store.GroupBusiness == nil { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"}) return } spec := &internalQuerySpec{ Select: `SELECT gm.id,gm.topic,gm.user_id,u.state,u.public,p.nickname,p.public_account,s.modewant AS mode_want,s.modegiven AS mode_given,s.recvseqid AS recv_seq_id,s.readseqid AS read_seq_id,s.delid AS del_id, gm.role,gm.member_status,gm.nickname AS group_nickname,gm.pinned,gm.mute_notifications,gm.join_source,gm.joined_by,gm.joined_at,gm.left_at,gm.removed_by,gm.version,gm.created_at,gm.updated_at`, From: "FROM im_group_members AS gm LEFT JOIN im_users AS u ON u.id=gm.user_id LEFT JOIN im_user_profiles AS p ON p.user_id=gm.user_id LEFT JOIN im_subscriptions AS s ON s.topic=gm.topic AND s.userid=gm.user_id", DefaultSort: "ORDER BY gm.id DESC", ScopeColumn: "gm.topic", Filters: []internalQueryFilter{ {"user_id", "gm.user_id", "=", "uid"}, {"role", "gm.role", "=", "string"}, {"member_status", "gm.member_status", "=", "string"}, {"q", "gm.nickname", "LIKE", "string"}, {"eligible_role", "gm.role", "=", "string"}, }, } query, countQuery, args, page, perPage, err := buildInternalAdminListQuery(req, spec, topic) if err == nil { if result, queryErr := store.InternalAdminQueries.Query(tenantID, query, countQuery, args, page, perPage); queryErr == nil { writeIMSuccess(wrt, req, map[string]any{ "items": normalizeInternalAdminRowsForAPI(result.Items), "total": result.Total, "page": result.Page, "per_page": result.PerPage, "next_cursor": "", "has_more": false, }) return } } limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 200) members, nextCursor, err := store.GroupBusiness.ListMembers(tenantID, topic, types.GroupMemberListOptions{ Role: strings.TrimSpace(req.URL.Query().Get("role")), EligibleRole: strings.TrimSpace(req.URL.Query().Get("eligible_role")), Query: strings.TrimSpace(req.URL.Query().Get("q")), Cursor: strings.TrimSpace(req.URL.Query().Get("cursor")), Limit: limit, }) if err != nil { writeInternalQueryError(wrt, req, err) return } items := make([]any, 0, len(members)) for _, member := range members { items = append(items, map[string]any{ "user_id": internalAdminUIDRaw(member.UserID), "role": member.Role, "member_status": member.MemberStatus, "nickname": member.Nickname, "pinned": member.Pinned, "mute_notifications": member.MuteNotifications, "join_source": member.JoinSource, "joined_by": internalAdminUIDRaw(member.JoinedBy), "joined_at": member.JoinedAt, "left_at": member.LeftAt, "removed_by": internalAdminUIDRaw(member.RemovedBy), "version": member.Version, "created_at": member.CreatedAt, "updated_at": member.UpdatedAt, }) } writeIMSuccess(wrt, req, map[string]any{"items": items, "next_cursor": nextCursor}) } func serveInternalMessageQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic, seqRaw string) { seq, err := strconv.ParseInt(seqRaw, 10, 64) if topic == "" || err != nil || seq <= 0 { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) return } resource := "private-messages" spec := privateMessagesInternalQuerySpec() where := "m.topic=? AND m.seqid=?" if strings.HasPrefix(topic, "grp") { resource = "group-messages" spec, err = groupMessagesInternalQuerySpecForTenant(tenantID) if err != nil { writeInternalQueryError(wrt, req, err) return } } spec.Select += ",m.head,JSON_UNQUOTE(m.content) AS content" row, err := store.InternalAdminQueries.QueryOne(tenantID, spec.Select+" "+spec.From+" WHERE "+where, []any{topic, seq}) if err != nil { writeInternalQueryError(wrt, req, err) return } msg, err := store.Messages.GetOne(tenantID, topic, seq) if err != nil { writeInternalQueryError(wrt, req, err) return } row["content"] = msg.Content row["content_excerpt"] = internalMessageExcerpt(msg.Content) row["conversation_type"] = strings.TrimSuffix(resource, "-messages") writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row)) } func serveInternalFileQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, fileID string) { if fileID == "" { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "file_id"}) return } fileDBID, ok := parseInternalAdminDBID(fileID) if !ok { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "file_id"}) return } spec, _ := internalAdminQuerySpec("files") row, err := store.InternalAdminQueries.QueryOne(tenantID, spec.Select+" "+spec.From+" WHERE id=?", []any{fileDBID}) if err != nil { writeInternalQueryError(wrt, req, err) return } writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row)) } func serveInternalKafkaDeadLettersQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) { if store.HTTPInfrastructure == nil { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 200) page, err := store.HTTPInfrastructure.ListKafkaDeadLetters(tenantID, types.KafkaDeadLetterListOptions{ Status: strings.TrimSpace(req.URL.Query().Get("status")), BizType: strings.TrimSpace(req.URL.Query().Get("biz_type")), BizID: strings.TrimSpace(req.URL.Query().Get("biz_id")), EventType: strings.TrimSpace(req.URL.Query().Get("event_type")), Cursor: strings.TrimSpace(req.URL.Query().Get("cursor")), Limit: limit, }) if err != nil { writeInternalQueryError(wrt, req, err) return } items := make([]any, 0, len(page.Items)) for _, dead := range page.Items { items = append(items, internalKafkaDeadLetterPayload(&dead, false)) } writeIMSuccess(wrt, req, map[string]any{ "items": items, "next_cursor": page.NextCursor, "has_more": page.NextCursor != "", }) } func serveInternalKafkaDeadLetterQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, deadID string) { if store.HTTPInfrastructure == nil || strings.TrimSpace(deadID) == "" { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "dead_id"}) return } dead, err := store.HTTPInfrastructure.GetKafkaDeadLetter(tenantID, deadID) if err != nil { writeInternalQueryError(wrt, req, err) return } writeIMSuccess(wrt, req, internalKafkaDeadLetterPayload(dead, true)) } func internalGroupSettingsPayload(settings *types.GroupSettings) map[string]any { if settings == nil { return nil } return map[string]any{ "topic": settings.Topic, "owner_user_id": internalAdminUIDRaw(settings.OwnerUserID), "name": settings.Name, "avatar_file_id": settings.AvatarFileID, "description": settings.Description, "group_status": settings.GroupStatus, "join_approval": settings.JoinApproval, "mute_all": settings.MuteAll, "member_upload_files": settings.MemberUploadFiles, "member_private_chat": settings.MemberPrivateChat, "member_invite": settings.MemberInvite, "max_member_count": settings.MaxMemberCount, "max_admin_count": settings.MaxAdminCount, "member_count": settings.MemberCount, "version": settings.Version, "created_by": internalAdminUIDRaw(settings.CreatedBy), "dissolved_by": internalAdminUIDRaw(settings.DissolvedBy), "dissolved_at": settings.DissolvedAt, "created_at": settings.CreatedAt, "updated_at": settings.UpdatedAt, } } func internalKafkaDeadLetterPayload(dead *types.KafkaDeadLetter, includeEnvelope bool) map[string]any { if dead == nil { return nil } payload := map[string]any{ "id": dead.ID, "dead_id": dead.DeadID, "event_id": dead.EventID, "parent_event_id": dead.ParentEventID, "source_topic": dead.SourceTopic, "consumer_group": dead.ConsumerGroup, "event_type": dead.EventType, "biz_type": dead.BizType, "biz_id": dead.BizID, "retry_count": dead.RetryCount, "last_reason": dead.LastReason, "last_error": dead.LastError, "status": dead.Status, "replay_count": dead.ReplayCount, "created_at": dead.CreatedAt, "updated_at": dead.UpdatedAt, } if includeEnvelope { var envelope any if len(dead.Envelope) > 0 && json.Unmarshal(dead.Envelope, &envelope) == nil { payload["envelope"] = envelope } else { payload["envelope_raw"] = string(dead.Envelope) } } return payload } func writeInternalQueryError(wrt http.ResponseWriter, req *http.Request, err error) { logs.Warn.Printf("internal admin query failed: method=%s path=%s err=%v", req.Method, req.URL.RequestURI(), err) switch { case errors.Is(err, types.ErrUserNotFound), errors.Is(err, types.ErrTopicNotFound), errors.Is(err, types.ErrNotFound): writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil) case errors.Is(err, types.ErrMalformed): writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) case errors.Is(err, types.ErrPermissionDenied): writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil) default: writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) } } func parseInternalUid(raw string) types.Uid { if uid := types.ParseUserId(raw); !uid.IsZero() { return uid } id, ok := parseInternalAdminDBID(raw) if !ok { return types.ZeroUid } return store.EncodeUid(id) } func internalServiceAllows(service internalAPIServiceConfig, capability string) bool { for _, allowed := range service.Allow { allowed = strings.TrimSpace(allowed) if allowed == "*" || allowed == capability { return true } } return false } func hashInternalRequestBody(req *http.Request) (string, error) { if req.Body == nil { sum := sha256.Sum256(nil) return hex.EncodeToString(sum[:]), nil } body, err := io.ReadAll(io.LimitReader(req.Body, 1<<20)) if err != nil { return "", err } if err := req.Body.Close(); err != nil { return "", err } req.Body = io.NopCloser(bytes.NewReader(body)) sum := sha256.Sum256(body) return hex.EncodeToString(sum[:]), nil } func verifyInternalSignature(secret, signingText, signature string) bool { signature = strings.TrimPrefix(signature, "sha256=") mac := hmac.New(sha256.New, []byte(secret)) _, _ = mac.Write([]byte(signingText)) expected := mac.Sum(nil) if decoded, err := hex.DecodeString(signature); err == nil { return hmac.Equal(decoded, expected) } if decoded, err := base64.StdEncoding.DecodeString(signature); err == nil { return hmac.Equal(decoded, expected) } return false } func claimInternalNonce(serviceName, nonce, requestClaim string, expiresAt time.Time) bool { ttl := time.Until(expiresAt) nonceKey := fmt.Sprintf("%s:%s", serviceName, nonce) if claimed, _, enabled, err := store.ClaimShortStateValue("internal_api_nonce", nonceKey, requestClaim, ttl); enabled { if err != nil { logs.Warn.Println("internal api: failed to claim nonce in redis", err) return false } return claimed } now := time.Now() internalAPI.nonces.Range(func(key, value any) bool { if claim, ok := value.(internalNonceClaim); ok && claim.expiresAt.Before(now) { internalAPI.nonces.Delete(key) } return true }) if _, loaded := internalAPI.nonces.LoadOrStore(nonceKey, internalNonceClaim{value: requestClaim, expiresAt: expiresAt}); loaded { return false } return true } type internalNonceClaim struct { value string expiresAt time.Time } func internalRequestClaimValue(signingText string) string { sum := sha256.Sum256([]byte(signingText)) return hex.EncodeToString(sum[:]) }