fix: harden identity, live access, drafts and setup (review only)

Source-reviewed hardening changes and regression tests. UNBUILT / UNTESTED: not a deployable release. Go formatting, compilation, automated regression tests, full frontend validation and staging acceptance remain required. See docs/hardening/VALIDATION.md. Main remains the original clean snapshot; no upstream Git history imported.
2026-10-16 12:15:20 +00:00
parent e57610a27c
commit 63e0457886
86 changed files with 5174 additions and 1929 deletions
--- a/internal/services/ws_service.go
+++ b/internal/services/ws_service.go
@@ -1,3 +1,5 @@
+// Modified in the independent hardening fork; see docs/hardening.
+
 package services

 import (
@@ -31,6 +33,7 @@ type wsService struct {
 	upgrader websocket.Upgrader
 	seq      atomic.Uint64
 	manager  *WsConnectionManager
+	authCheckPeriod time.Duration
 }

 func newWsService() *wsService {
@@ -41,6 +44,7 @@ func newWsService() *wsService {
 			},
 		},
 		manager: newWsConnectionManager(),
+		authCheckPeriod: realtimeAuthCheckPeriod,
 	}
 }

@@ -50,6 +54,10 @@ func (s *wsService) HandleDashboardWS(ctx *gin.Context) {
 		ctx.AbortWithStatusJSON(http.StatusUnauthorized, web.JsonErrorCode(errorsx.CodeAuthUnauthorized, i18nx.T(ctx, "error.auth.expired")))
 		return
 	}
+ if !slices.Contains(principal.Permissions, constants.PermissionConversationView.Code) {
+  ctx.AbortWithStatusJSON(http.StatusForbidden, web.JsonErrorCode(errorsx.CodeAuthForbidden, i18nx.T(ctx, "error.e0225")))
+  return
+ }
 	if err := s.upgradeConnection(ctx, principal, nil, realtimeRoleAdmin); err != nil {
 		slog.Error("upgrade admin websocket failed", "error", err, "path", ctx.Request.URL.Path)
 		ctx.Abort()
@@ -77,29 +85,35 @@ func (s *wsService) HandleOpenWS(ctx *gin.Context) {
 		return
 	}

-	var (
-		principal           = AuthService.GetAuthPrincipal(ctx)
-		external            *openidentity.ExternalUser
-		customerSessionInfo *CustomerSessionVerifyResult
-	)
-	if principal == nil {
-		result, err := CustomerSessionService.VerifyRequest(ctx, channel)
-		if err != nil {
-			ctx.AbortWithStatusJSON(http.StatusUnauthorized, web.JsonError(err))
-			return
-		}
-		external = result.ExternalUser
-		customerSessionInfo = result
-	}
-	if err := s.upgradeConnection(ctx, principal, external, realtimeRoleUser, customerSessionInfo); err != nil {
-		slog.Error("upgrade open im websocket failed", "error", err, "path", ctx.Request.URL.Path, "channelId", channel.ChannelID, "channel_id", channel.ID)
-		ctx.Abort()
-		return
-	}
+ // The customer endpoint always authenticates the customer credential. A staff
+ // cookie/bearer must not turn this into a second dashboard subscription path.
+ result, err := CustomerSessionService.VerifyRequest(ctx, channel)
+ if err != nil {
+  ctx.AbortWithStatusJSON(http.StatusUnauthorized, web.JsonError(err))
+  return
+ }
+ if err := s.upgradeConnection(ctx, nil, result.ExternalUser, realtimeRoleUser, result); err != nil {
+  slog.Error("upgrade open im websocket failed", "error", err, "path", ctx.Request.URL.Path, "channel_id", channel.ID)
+  ctx.Abort()
+  return
+ }
 }

 func (s *wsService) upgradeConnection(ctx *gin.Context, principal *dto.AuthPrincipal, external *openidentity.ExternalUser, role string, customerSessionInfo ...*CustomerSessionVerifyResult) error {
-	conn, err := s.upgrader.Upgrade(ctx.Writer, ctx.Request, nil)
+ if principal != nil {
+  fresh, err := AuthService.RevalidatePrincipal(principal)
+  if err != nil {
+   ctx.AbortWithStatusJSON(http.StatusUnauthorized, web.JsonError(err))
+   return err
+  }
+  principal = fresh
+  if role == realtimeRoleAdmin && !slices.Contains(principal.Permissions, constants.PermissionConversationView.Code) {
+   err := errorsx.ForbiddenI18n("error.e0225")
+   ctx.AbortWithStatusJSON(http.StatusForbidden, web.JsonError(err))
+   return err
+  }
+ }
+ conn, err := s.upgrader.Upgrade(ctx.Writer, ctx.Request, nil)
 	if err != nil {
 		return err
 	}
@@ -114,9 +128,10 @@ func (s *wsService) upgradeConnection(ctx *gin.Context, principal *dto.AuthPrinc
 		Topics:       make(map[string]struct{}),
 		Send:         make(chan []byte, realtimeSendBufferSize),
 	}
-	session.touch()
+ if len(customerSessionInfo) > 0 { session.CustomerSession = customerSessionInfo[0] }
+ session.touch()

-	conn.SetReadLimit(realtimeMaxMessageSize)
+ conn.SetReadLimit(realtimeMaxMessageSize)
 	_ = conn.SetReadDeadline(time.Now().Add(realtimePongWait))
 	conn.SetPongHandler(func(string) error {
 		session.touch()
@@ -139,13 +154,10 @@ func (s *wsService) upgradeConnection(ctx *gin.Context, principal *dto.AuthPrinc
 		"userId", logUserID,
 		"externalId", logExternalID,
 		"terminalType", session.TerminalType,
-		"topicCount", len(session.Topics),
+		"topicCount", len(s.manager.TopicList(session)),
 		"sessionCount", sessionCount,
 	)

-	go s.writePump(session)
-	go s.readPump(session)
-
 	session.enqueueEvent(s.newEvent("", RealtimeConnectedEvent{
 		Payload: RealtimeConnectedPayload{
 			ConnID:       session.ID,
@@ -153,7 +165,7 @@ func (s *wsService) upgradeConnection(ctx *gin.Context, principal *dto.AuthPrinc
 			GuestID:      logExternalID,
 			Role:         role,
 			TerminalType: session.TerminalType,
-			Topics:       session.topicList(),
+			Topics:       s.manager.TopicList(session),
 		},
 	}))
 	if len(customerSessionInfo) > 0 && customerSessionInfo[0] != nil && customerSessionInfo[0].Refreshed {
@@ -164,6 +176,8 @@ func (s *wsService) upgradeConnection(ctx *gin.Context, principal *dto.AuthPrinc
 			},
 		}))
 	}
+	go s.writePump(session)
+	go s.readPump(session)
 	return nil
 }

@@ -181,6 +195,7 @@ func (s *wsService) readPump(session *ClientSession) {
 }

 func (s *wsService) handleClientMessage(session *ClientSession, payload []byte) {
+	if !s.revalidateSession(session) { return }
 	input := realtimeClientMessage{}
 	if err := json.Unmarshal(payload, &input); err != nil {
 		session.enqueueEvent(s.newEvent("", RealtimeResyncRequiredEvent{
@@ -214,7 +229,7 @@ func (s *wsService) handleClientMessage(session *ClientSession, payload []byte)
 			"eventId", strings.TrimSpace(input.EventID),
 		)
 	case enums.IMRealtimeClientTypePresence:
-		if input.ConversationID > 0 && session.Principal != nil {
+		if input.ConversationID > 0 && session.Role == realtimeRoleAdmin && session.Principal != nil && s.canSubscribeConversation(session, input.ConversationID) {
 			action := strings.TrimSpace(input.Action)
 			if action == "" {
 				action = "viewing"
@@ -232,8 +247,10 @@ func (s *wsService) handleClientMessage(session *ClientSession, payload []byte)

 func (s *wsService) writePump(session *ClientSession) {
 	ticker := time.NewTicker(realtimePingPeriod)
+	authTicker := time.NewTicker(s.authCheckPeriod)
 	defer func() {
 		ticker.Stop()
+		authTicker.Stop()
 		s.closeSession(session)
 	}()

@@ -245,9 +262,16 @@ func (s *wsService) writePump(session *ClientSession) {
 				_ = session.Conn.WriteMessage(websocket.CloseMessage, []byte{})
 				return
 			}
-			if err := session.Conn.WriteMessage(websocket.TextMessage, payload); err != nil {
+   // Recheck after queueing as well: logout or permission revocation can
+   // happen while a slow client still has messages buffered.
+   if !s.revalidateSession(session) { return }
+   var envelope struct { Topic string `json:"topic"` }
+   if json.Unmarshal(payload, &envelope) != nil || (envelope.Topic != "" && len(s.filterAllowedTopics(session, []string{envelope.Topic})) == 0) { continue }
+   if err := session.Conn.WriteMessage(websocket.TextMessage, payload); err != nil {
 				return
 			}
+		case <-authTicker.C:
+			if !s.revalidateSession(session) { return }
 		case <-ticker.C:
 			_ = session.Conn.SetWriteDeadline(time.Now().Add(realtimeWriteWait))
 			if err := session.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
@@ -262,10 +286,12 @@ func (s *wsService) closeSession(session *ClientSession) {
 		return
 	}
 	session.closeOnce.Do(func() {
+		session.sendMu.Lock()
 		session.Closed.Store(true)
+		if session.Send != nil { close(session.Send) }
+		session.sendMu.Unlock()
 		remaining := s.manager.Unregister(session)

-		close(session.Send)
 		if session.Conn != nil {
 			_ = session.Conn.Close()
 		}
@@ -586,8 +612,9 @@ func (s *wsService) PublishToTopicsWithFilter(topics []string, event RealtimeEve
 		return
 	}

-	for _, session := range targets {
-		if filter != nil && !filter(session) {
+ for _, session := range targets {
+  if !s.revalidateSession(session) || !s.canReceiveEvent(session, normalized, event.Topic) { continue }
+  if filter != nil && !filter(session) {
 			continue
 		}
 		if session.enqueue(payload) {
@@ -603,12 +630,9 @@ func (s *wsService) PublishToTopicsWithFilter(topics []string, event RealtimeEve
 	}
 }

-func (s *wsService) IsGuestOnline(guestID string) bool {
-	guestID = strings.TrimSpace(guestID)
-	if guestID == "" {
-		return false
-	}
-	return s.manager.HasTopic(s.guestTopic(guestID))
+func (s *wsService) IsCustomerOnline(customerID, channelID int64) bool {
+ if customerID <= 0 || channelID <= 0 { return false }
+ return s.manager.HasTopic(s.customerTopic(customerID, channelID))
 }

 func (s *wsService) routeConversationTopics(conversation *models.Conversation) []string {
@@ -617,8 +641,8 @@ func (s *wsService) routeConversationTopics(conversation *models.Conversation) [
 	}

 	topics := []string{s.conversationTopic(conversation.ID)}
-	if identity := ConversationService.GetConversationExternalIdentity(conversation); identity != nil && strings.TrimSpace(identity.ExternalID) != "" {
-		topics = append(topics, s.guestTopic(identity.ExternalID))
+	if conversation.CustomerID > 0 && conversation.ChannelID > 0 {
+		topics = append(topics, s.customerTopic(conversation.CustomerID, conversation.ChannelID))
 	}
 	if conversation.CurrentAssigneeID > 0 {
 		topics = append(topics, s.adminTopic(conversation.CurrentAssigneeID))
@@ -640,19 +664,15 @@ func (s *wsService) defaultTopics(session *ClientSession) []string {
 		}
 		return []string{s.notificationTopic(session.Principal.UserID)}
 	case realtimeRoleAdmin:
-		if session.Principal == nil || session.Principal.UserID <= 0 {
-			return []string{realtimeTopicAdminAll}
+		if session.Principal == nil || session.Principal.UserID <= 0 || !slices.Contains(session.Principal.Permissions, constants.PermissionConversationView.Code) {
+			return nil
 		}
 		return []string{s.adminTopic(session.Principal.UserID), realtimeTopicAdminAll}
-	default:
-		// 开放 IM：仅 External、无 AuthPrincipal 的访客连接必须仍能订阅 guest:{externalId}，否则收不到推送。
-		if session.External != nil && strings.TrimSpace(session.External.ExternalID) != "" {
-			return []string{s.guestTopic(session.External.ExternalID)}
-		}
-		if session.Principal != nil && session.Principal.UserID > 0 {
-			return []string{s.userTopic(session.Principal.UserID)}
-		}
-		return nil
+ default:
+  if session.CustomerSession != nil && session.External != nil && session.CustomerSession.CustomerID > 0 && session.CustomerSession.ChannelID > 0 {
+   return []string{s.customerTopic(session.CustomerSession.CustomerID, session.CustomerSession.ChannelID)}
+  }
+  return nil
 	}
 }

@@ -693,7 +713,7 @@ func (s *wsService) filterAllowedTopics(session *ClientSession, topics []string)
 }

 func (s *wsService) canSubscribeConversation(session *ClientSession, conversationID int64) bool {
-	if session == nil || conversationID <= 0 {
+	if session == nil || conversationID <= 0 || session.Role == realtimeRoleNotification {
 		return false
 	}
 	if session.Role == realtimeRoleAdmin {
@@ -751,12 +771,8 @@ func (s *wsService) nextID(prefix string) string {
 	return fmt.Sprintf("%s_%d_%d", prefix, time.Now().UnixNano(), seq)
 }

-func (s *wsService) userTopic(userID int64) string {
-	return realtimeTopicUserPrefix + strconv.FormatInt(userID, 10)
-}
-
-func (s *wsService) guestTopic(guestID string) string {
-	return realtimeTopicGuestPrefix + strings.TrimSpace(guestID)
+func (s *wsService) customerTopic(customerID, channelID int64) string {
+ return realtimeTopicCustomerPrefix + strconv.FormatInt(channelID, 10) + ":" + strconv.FormatInt(customerID, 10)
 }

 func (s *wsService) adminTopic(userID int64) string {
@@ -817,3 +833,24 @@ func formatWsTime(t *time.Time) string {
 	}
 	return t.Format(time.DateTime)
 }
+
+// Topic membership is an optimization, never an authorization grant.
+func (s *wsService) canReceiveEvent(session *ClientSession, routingTopics []string, eventTopic string) bool {
+ if eventTopic != "" { return len(s.filterAllowedTopics(session, []string{eventTopic})) > 0 }
+ return len(s.filterAllowedTopics(session, routingTopics)) > 0
+}
+
+// Principal snapshots are immutable. Closing when their scope changes avoids
+// races between fanout/read/write goroutines and forces a fresh subscription set.
+func (s *wsService) revalidateSession(session *ClientSession) bool {
+ if session == nil || session.Closed.Load() { return false }
+ valid := false
+ if session.Principal != nil {
+  fresh, err := AuthService.RevalidatePrincipal(session.Principal)
+  valid = err == nil && fresh.UserType == session.Principal.UserType && slices.Equal(fresh.Roles, session.Principal.Roles) && slices.Equal(fresh.Permissions, session.Principal.Permissions)
+ } else if session.Role == realtimeRoleUser {
+  valid = CustomerSessionService.VerifyConnection(session.CustomerSession)
+ }
+ if !valid { s.closeSession(session) }
+ return valid
+}