Fix race condition, error normalization, and security concerns

- Fix startForwarding race: capture upstream under lock before goroutine start
- Reject SUBSCRIBE/PSUBSCRIBE during MULTI transaction to prevent state corruption
- Normalize non-redis.Error to "ERR ..." prefix in writeRedisError for valid RESP
- Truncate divergence values in Sentry reports to prevent data leakage

Author: bootjp

proxy/proxy.go

@@ -378,12 +378,16 @@ func writeResponse(w respWriter, resp any, err error) {
 	writeRedisValue(w, resp)
 }
 
-// writeRedisError writes an upstream error without double-prefixing.
-// Redis errors already contain their prefix (e.g. "ERR ...", "WRONGTYPE ...").
+// writeRedisError writes an upstream error to the client.
+// go-redis redis.Error values already carry the Redis prefix (e.g. "ERR ...", "WRONGTYPE ...").
+// Other errors (timeouts, dial failures) are normalized to "ERR ..." to produce valid RESP.
 func writeRedisError(w respWriter, err error) {
-	msg := err.Error()
-	// go-redis errors are already formatted with prefix; pass through as-is.
-	w.WriteError(msg)
+	var redisErr redis.Error
+	if errors.As(err, &redisErr) {
+		w.WriteError(redisErr.Error())
+		return
+	}
+	w.WriteError("ERR " + err.Error())
 }
 
 // writeRedisValue writes a go-redis response value to a respWriter.

proxy/pubsub.go

@@ -75,17 +75,24 @@ func (s *pubsubSession) cleanup() {
 }
 
 func (s *pubsubSession) startForwarding() {
+	// Capture upstream under lock to avoid race with exitPubSubMode.
+	s.mu.Lock()
+	upstream := s.upstream
+	s.mu.Unlock()
+	if upstream == nil {
+		return
+	}
+	ch := upstream.Channel()
 	s.fwdDone = make(chan struct{})
 	go func() {
 		defer close(s.fwdDone)
-		s.forwardMessages()
+		s.forwardMessages(ch)
 	}()
 }
 
 // forwardMessages reads from the upstream go-redis PubSub channel and writes
 // messages to the detached client connection.
-func (s *pubsubSession) forwardMessages() {
-	ch := s.upstream.Channel()
+func (s *pubsubSession) forwardMessages(ch <-chan *redis.Message) {
 	for msg := range ch {
 		s.mu.Lock()
 		if s.closed {
@@ -198,6 +205,10 @@ func (s *pubsubSession) dispatchNormalCommand(name string, args [][]byte) bool {
 		return true
 	}
 	if name == cmdSubscribe || name == cmdPSubscribe {
+		if s.inTxn {
+			s.writeError("ERR Command not allowed inside a transaction")
+			return true
+		}
 		s.reenterPubSub(name, args)
 		return true
 	}

proxy/sentry.go

@@ -11,6 +11,9 @@ import (
 
 const (
 	defaultReportCooldown = 60 * time.Second
+	// maxSentryValueLen limits the length of values attached to Sentry events
+	// to prevent data leakage and oversized events.
+	maxSentryValueLen = 256
 	// maxReportEntries caps the lastReport map to prevent unbounded growth.
 	maxReportEntries = 10000
 )
@@ -80,8 +83,8 @@ func (r *SentryReporter) CaptureDivergence(div Divergence) {
 		scope.SetTag("command", div.Command)
 		scope.SetTag("key", div.Key)
 		scope.SetTag("kind", div.Kind.String())
-		scope.SetExtra("primary", fmt.Sprintf("%v", div.Primary))
-		scope.SetExtra("secondary", fmt.Sprintf("%v", div.Secondary))
+		scope.SetExtra("primary", truncateValue(div.Primary))
+		scope.SetExtra("secondary", truncateValue(div.Secondary))
 		scope.SetFingerprint([]string{"divergence", div.Kind.String(), div.Command})
 		scope.SetLevel(sentry.LevelWarning)
 		r.hub.CaptureMessage(fmt.Sprintf("data divergence: %s %s (%s)", div.Kind, div.Command, div.Key))
@@ -129,3 +132,12 @@ func cmdNameFromArgs(args [][]byte) string {
 	}
 	return unknownStr
 }
+
+// truncateValue formats a value for Sentry, truncating to avoid data leakage and oversized events.
+func truncateValue(v any) string {
+	s := fmt.Sprintf("%v", v)
+	if len(s) > maxSentryValueLen {
+		return s[:maxSentryValueLen] + "...(truncated)"
+	}
+	return s
+}