diff --git a/server/im_http.go b/server/im_http.go index 3b7e6b4..a29ac58 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -5944,17 +5944,40 @@ func serveIMGroupSummary(wrt http.ResponseWriter, req *http.Request, topic strin writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil) return } - principal, selfSub, ok := requireIMGroupMember(wrt, req, topic, true) + principal, ok := authenticateIMHTTPBearer(wrt, req) if !ok { return } + if types.GetTopicCat(topic) != types.TopicCatGrp { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"}) + return + } + // 已解散的群按文档约定对客户端表现为不存在（RESOURCE_NOT_FOUND），不暴露解散状态。 + if settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic); err == nil && + settings != nil && settings.GroupStatus == types.GroupStatusDissolved { + writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonResourceNotFound, "group not found", nil) + return + } + selfSub, err := store.Subs.Get(principal.TenantID, topic, principal.Uid, false) + if err != nil { + writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) + return + } + access := types.ModeNone + if selfSub != nil { + access = selfSub.ModeGiven & selfSub.ModeWant + } + if selfSub == nil || !access.IsJoiner() || !access.IsReader() { + writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "topic permission required", nil) + return + } requests, _, _ := store.GroupBusiness.ListJoinRequests(principal.TenantID, topic, types.GroupJoinRequestStatusPending, "", 100) announcement, _ := store.GroupBusiness.LatestAnnouncement(principal.TenantID, topic) writeIMSuccess(wrt, req, map[string]any{ "topic": topic, "pending_join_request_count": len(requests), "announcement": groupAnnouncementPayload(announcement), - "permissions": groupPermissions(selfSub.ModeGiven & selfSub.ModeWant), + "permissions": groupPermissions(access), }) } diff --git a/server/session.go b/server/session.go index 9d78ce6..e60c023 100644 --- a/server/session.go +++ b/server/session.go @@ -749,7 +749,14 @@ func (s *Session) subscribe(msg *ClientComMessage) { s.inflightReqs.Add(1) // Session can subscribe to topic on behalf of a single user at a time. - if sub := s.getSub(msg.RcptTo); sub != nil { + // Fast-path the duplicate attach only when the client used the canonical + // topic name itself. A p2p topic may also be addressed by an alias (the + // peer's user id, e.g. "usrXXX") while the session is already attached to + // the canonical topic ("p2pXXX"): forwarding it to the topic lets the + // client complete the alias->canonical rename with a 200 reply instead of + // a 304 that would leave the client-side alias topic object unattached + // (publishes on it then fail with "Cannot publish on inactive topic"). + if sub := s.getSub(msg.RcptTo); sub != nil && msg.Original == msg.RcptTo { s.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp)) s.inflightReqs.Done() } else { diff --git a/server/topic.go b/server/topic.go index 6db9a76..bf87d6e 100644 --- a/server/topic.go +++ b/server/topic.go @@ -353,7 +353,30 @@ func (t *Topic) registerSession(msg *ClientComMessage) { // but there is a gap between topic creation/un-pausing and processing the // first subscription request, before the topic is linked to session: a client // may send several subscription requests in that gap. - msg.sess.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp)) + // + // A p2p topic may be addressed by an alias (the peer's user id, e.g. "usrXXX") + // while the session is already attached to the same topic under its canonical + // name ("p2pXXX"). Replying 304 here leaves the client-side topic object created + // for the alias name unattached: the SDK treats any ctrl code >= 300 as failure + // and skips the alias->canonical rename, so publishes on it fail with + // "Cannot publish on inactive topic". Reply 200 with the canonical name and the + // effective access mode so the client can complete the rename instead. + asUid := types.ParseUserId(msg.AsUser) + if pud, ok := t.perUser[asUid]; ok && t.cat == types.TopicCatP2P && msg.Original != t.name { + msg.sess.queueOut(NoErrParams(msg.Id, t.name, types.TimeNow(), map[string]any{ + "acs": &MsgAccessMode{ + Want: pud.modeWant.String(), + Given: pud.modeGiven.String(), + Mode: (pud.modeGiven & pud.modeWant).String(), + }, + })) + // The client-side topic object created for the alias name is fresh: + // serve the {get} query it sent along with the subscription, otherwise + // the reopened conversation would come up without desc/subs/history. + t.processSubGetQuery(msg, asUid, false, auth.Level(msg.AuthLvl)) + } else { + msg.sess.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp)) + } } else { // The topic is alive, so stop the kill timer, if it's ticking. We don't want the topic to die // while processing the call. @@ -879,60 +902,71 @@ func (t *Topic) handleSubscription(msg *ClientComMessage) error { } msgsub := msg.Sub - getWhat := 0 if msgsub.Get != nil { - getWhat = parseMsgClientMeta(msgsub.Get.What) + // Deliver the data requested by the {get} query attached to the subscription. + t.processSubGetQuery(msg, asUid, asChan, authLevel) } + + return nil +} + +// processSubGetQuery responds to the {get} query attached to a subscription +// request (desc/sub/tags/cred/aux/data/del) by sending the answers as separate +// {meta}/{data} packets. +func (t *Topic) processSubGetQuery(msg *ClientComMessage, asUid types.Uid, asChan bool, authLevel auth.Level) { + msgsub := msg.Sub + if msgsub == nil || msgsub.Get == nil { + return + } + getWhat := parseMsgClientMeta(msgsub.Get.What) if getWhat&constMsgMetaDesc != 0 { // Send get.desc as a {meta} packet. if err := t.replyGetDesc(msg.sess, asUid, asChan, msgsub.Get.Desc, msg); err != nil { - logs.Warn.Printf("topic[%s] handleSubscription Get.Desc failed: %v sid=%s", t.name, err, msg.sess.sid) + logs.Warn.Printf("topic[%s] Get.Desc failed: %v sid=%s", t.name, err, msg.sess.sid) } } if getWhat&constMsgMetaSub != 0 { // Send get.sub response as a separate {meta} packet if err := t.replyGetSub(msg.sess, asUid, authLevel, asChan, msg); err != nil { - logs.Warn.Printf("topic[%s] handleSubscription Get.Sub failed: %v sid=%s", t.name, err, msg.sess.sid) + logs.Warn.Printf("topic[%s] Get.Sub failed: %v sid=%s", t.name, err, msg.sess.sid) } } if getWhat&constMsgMetaTags != 0 { // Send get.tags response as a separate {meta} packet if err := t.replyGetTags(msg.sess, asUid, msg); err != nil { - logs.Warn.Printf("topic[%s] handleSubscription Get.Tags failed: %v sid=%s", t.name, err, msg.sess.sid) + logs.Warn.Printf("topic[%s] Get.Tags failed: %v sid=%s", t.name, err, msg.sess.sid) } } if getWhat&constMsgMetaCred != 0 { - // Send get.tags response as a separate {meta} packet + // Send get.cred response as a separate {meta} packet if err := t.replyGetCreds(msg.sess, asUid, msg); err != nil { - logs.Warn.Printf("topic[%s] handleSubscription Get.Cred failed: %v sid=%s", t.name, err, msg.sess.sid) + logs.Warn.Printf("topic[%s] Get.Cred failed: %v sid=%s", t.name, err, msg.sess.sid) } } if getWhat&constMsgMetaAux != 0 { // Send get.aux response as a separate {meta} packet if err := t.replyGetAux(msg.sess, asUid, msg); err != nil { - logs.Warn.Printf("topic[%s] handleSubscription Get.Aux failed: %v sid=%s", t.name, err, msg.sess.sid) + logs.Warn.Printf("topic[%s] Get.Aux failed: %v sid=%s", t.name, err, msg.sess.sid) } } if getWhat&constMsgMetaData != 0 { // Send get.data response as {data} packets if err := t.replyGetData(msg.sess, asUid, asChan, msgsub.Get.Data, msg); err != nil { - logs.Warn.Printf("topic[%s] handleSubscription Get.Data failed: %v sid=%s", t.name, err, msg.sess.sid) + logs.Warn.Printf("topic[%s] Get.Data failed: %v sid=%s", t.name, err, msg.sess.sid) } } if getWhat&constMsgMetaDel != 0 { // Send get.del response as a separate {meta} packet if err := t.replyGetDel(msg.sess, asUid, msgsub.Get.Del, msg); err != nil { - logs.Warn.Printf("topic[%s] handleSubscription Get.Del failed: %v sid=%s", t.name, err, msg.sess.sid) + logs.Warn.Printf("topic[%s] Get.Del failed: %v sid=%s", t.name, err, msg.sess.sid) } } - - return nil } // handleLeaveRequest processes a session leave request.