Review only: HOUSE-79 upstream v8.0.44, part 7 of 13: sdk handlers and api server

internal/api/middleware/request_logging.go

@@ -5,8 +5,10 @@ package middleware
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"os"
 	"strings"
@@ -52,6 +54,11 @@ func RequestLoggingMiddleware(logger logging.RequestLogger) gin.HandlerFunc {
 		// Capture request information
 		requestInfo, err := captureRequestInfo(c, captureBody)
 		if err != nil {
+			var maxBytesErr *http.MaxBytesError
+			if errors.As(err, &maxBytesErr) {
+				c.AbortWithStatusJSON(http.StatusRequestEntityTooLarge, gin.H{"error": err.Error()})
+				return
+			}
 			// Log error but continue processing
 			// In a real implementation, you might want to use a proper logger here
 			c.Next()
@@ -338,7 +345,12 @@ func captureRequestInfo(c *gin.Context, captureBody bool) (*RequestInfo, error)
 
 		// Restore the body for the actual request processing
 		c.Request.Body = io.NopCloser(bytes.NewBuffer(bodyBytes))
-		body = decodeCapturedRequestBodyForLog(bodyBytes, c.Request.Header.Get("Content-Encoding"))
+		maxBytes := c.GetInt64("maxRequestBodyBytes")
+		if maxBytes > 0 {
+			body = decodeCapturedRequestBodyForLogWithLimit(bodyBytes, c.Request.Header.Get("Content-Encoding"), maxBytes)
+		} else {
+			body = decodeCapturedRequestBodyForLog(bodyBytes, c.Request.Header.Get("Content-Encoding"))
+		}
 	}
 
 	return &RequestInfo{
@@ -445,7 +457,11 @@ func decodeCapturedZstdRequestBodyWithLimit(raw []byte, limit int64) ([]byte, bo
 	}
 	defer decoder.Close()
 
-	decoded, errRead := io.ReadAll(io.LimitReader(decoder, limit+1))
+	var reader io.Reader = decoder
+	if limit > 0 && limit < math.MaxInt64 {
+		reader = io.LimitReader(decoder, limit+1)
+	}
+	decoded, errRead := io.ReadAll(reader)
 	if errRead != nil {
 		return nil, false, fmt.Errorf("failed to decode zstd request body: %w", errRead)
 	}

internal/api/middleware/response_writer.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/router-for-me/CLIProxyAPI/v8/internal/clienterror"
+	"github.com/router-for-me/CLIProxyAPI/v8/internal/httpwire"
 	"github.com/router-for-me/CLIProxyAPI/v8/internal/interfaces"
 	"github.com/router-for-me/CLIProxyAPI/v8/internal/logging"
 	log "github.com/sirupsen/logrus"
@@ -104,6 +105,11 @@ func (w *ResponseWriterWrapper) Write(data []byte) (int, error) {
 	return n, err
 }
 
+// FlushError forwards transport errors without bypassing inner middleware.
+func (w *ResponseWriterWrapper) FlushError() error {
+	return httpwire.FlushResponse(w.ResponseWriter)
+}
+
 func (w *ResponseWriterWrapper) shouldBufferResponseBody() bool {
 	if w.logger != nil && w.logger.IsEnabled() {
 		return true

internal/api/protocol_multiplexer.go

@@ -60,6 +60,21 @@ func (s *Server) acceptMuxConnections(listener net.Listener, httpListener *muxLi
 
 // routeMuxConnection performs per-connection protocol detection and routing.
 func (s *Server) routeMuxConnection(conn net.Conn, httpListener *muxListener) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			var remoteAddr net.Addr
+			if conn != nil {
+				remoteAddr = conn.RemoteAddr()
+			}
+			log.Errorf("panic while routing connection from %v: %v", remoteAddr, recovered)
+			if conn != nil {
+				if errClose := conn.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) {
+					log.Errorf("failed to close connection after panic: %v", errClose)
+				}
+			}
+		}
+	}()
+
 	// Set a read deadline so that idle connections that never send bytes do not
 	// leak goroutines and file descriptors. The deadline is cleared once the
 	// connection is successfully routed to its handler.
@@ -82,12 +97,11 @@ func (s *Server) routeMuxConnection(conn net.Conn, httpListener *muxListener) {
 				}
 				return
 			}
+			_ = conn.SetReadDeadline(time.Time{})
 			if errPut := httpListener.Put(tlsConn); errPut != nil {
 				if errClose := conn.Close(); errClose != nil {
 					log.Errorf("failed to close connection after HTTP routing failure: %v", errClose)
 				}
-			} else {
-				_ = conn.SetReadDeadline(time.Time{})
 			}
 			return
 		}
@@ -115,11 +129,10 @@ func (s *Server) routeMuxConnection(conn net.Conn, httpListener *muxListener) {
 		return
 	}
 
+	_ = conn.SetReadDeadline(time.Time{})
 	if errPut := httpListener.Put(&bufferedConn{Conn: conn, reader: reader}); errPut != nil {
 		if errClose := conn.Close(); errClose != nil {
 			log.Errorf("failed to close connection after HTTP routing failure: %v", errClose)
 		}
-	} else {
-		_ = conn.SetReadDeadline(time.Time{})
 	}
 }

internal/api/redis_queue_protocol.go

@@ -17,6 +17,11 @@ import (
 const (
 	redisUsageChannel  = "usage"
 	redisErrorsChannel = "errors"
+
+	maxRESPArrayLen   = 1024
+	maxRESPBulkLen    = 64 << 10 // 64 KiB
+	maxRESPFrameBytes = 64 << 10 // 64 KiB
+	maxRESPLineLen    = 64 << 10 // 64 KiB
 )
 
 type redisSubscriptionCommand struct {
@@ -45,7 +50,10 @@ func (s *Server) handleRedisConnection(conn net.Conn, reader *bufio.Reader) {
 	authed := false
 	writer := bufio.NewWriter(conn)
 	defer func() {
-		if errClose := conn.Close(); errClose != nil {
+		if recovered := recover(); recovered != nil {
+			log.Errorf("panic while handling redis connection from %v: %v", conn.RemoteAddr(), recovered)
+		}
+		if errClose := conn.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) {
 			log.Errorf("redis connection close error: %v", errClose)
 		}
 	}()
@@ -58,7 +66,7 @@ func (s *Server) handleRedisConnection(conn net.Conn, reader *bufio.Reader) {
 		return true
 	}
 
-	if s.cfg != nil && s.cfg.Home.Enabled {
+	if cfg := s.getConfig(); cfg != nil && cfg.Home.Enabled {
 		_ = writeRedisError(writer, "ERR redis usage output disabled in home mode")
 		_ = writer.Flush()
 		return
@@ -289,6 +297,11 @@ func (s *Server) streamRedisSubscription(reader *bufio.Reader, writer *bufio.Wri
 
 func readRedisSubscriptionCommands(reader *bufio.Reader, commands chan<- redisSubscriptionCommand, done <-chan struct{}) {
 	defer close(commands)
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			log.Errorf("panic while reading redis subscription commands: %v", recovered)
+		}
+	}()
 
 	for {
 		args, errRead := readRESPArray(reader)
@@ -423,12 +436,13 @@ func readRESPArray(reader *bufio.Reader) ([]string, error) {
 		return nil, errLine
 	}
 	count, errParse := strconv.Atoi(line)
-	if errParse != nil || count < 0 {
+	if errParse != nil || count < 0 || count > maxRESPArrayLen {
 		return nil, fmt.Errorf("protocol error")
 	}
 	args := make([]string, 0, count)
+	remainingBytes := maxRESPFrameBytes
 	for i := 0; i < count; i++ {
-		value, errString := readRESPString(reader)
+		value, errString := readRESPStringWithBudget(reader, &remainingBytes)
 		if errString != nil {
 			return nil, errString
 		}
@@ -438,21 +452,39 @@ func readRESPArray(reader *bufio.Reader) ([]string, error) {
 }
 
 func readRESPString(reader *bufio.Reader) (string, error) {
+	return readRESPStringWithBudget(reader, nil)
+}
+
+func readRESPStringWithBudget(reader *bufio.Reader, remainingBytes *int) (string, error) {
 	prefix, errRead := reader.ReadByte()
 	if errRead != nil {
 		return "", errRead
 	}
 	switch prefix {
 	case '$':
-		return readRESPBulkString(reader)
+		return readRESPBulkStringWithBudget(reader, remainingBytes)
 	case '+', ':':
-		return readRESPLine(reader)
+		line, errLine := readRESPLine(reader)
+		if errLine != nil {
+			return "", errLine
+		}
+		if remainingBytes != nil {
+			if len(line) > *remainingBytes {
+				return "", fmt.Errorf("protocol error")
+			}
+			*remainingBytes -= len(line)
+		}
+		return line, nil
 	default:
 		return "", fmt.Errorf("protocol error")
 	}
 }
 
 func readRESPBulkString(reader *bufio.Reader) (string, error) {
+	return readRESPBulkStringWithBudget(reader, nil)
+}
+
+func readRESPBulkStringWithBudget(reader *bufio.Reader, remainingBytes *int) (string, error) {
 	line, errLine := readRESPLine(reader)
 	if errLine != nil {
 		return "", errLine
@@ -461,27 +493,56 @@ func readRESPBulkString(reader *bufio.Reader) (string, error) {
 	if errParse != nil {
 		return "", fmt.Errorf("protocol error")
 	}
+	if length < -1 {
+		return "", fmt.Errorf("protocol error")
+	}
 	if length < 0 {
 		return "", nil
 	}
+	if length > maxRESPBulkLen || (remainingBytes != nil && length > *remainingBytes) {
+		return "", fmt.Errorf("bulk string too large")
+	}
 	buf := make([]byte, length+2)
 	if _, errRead := io.ReadFull(reader, buf); errRead != nil {
 		return "", errRead
 	}
-	if length+2 < 2 || buf[length] != '\r' || buf[length+1] != '\n' {
+	if buf[length] != '\r' || buf[length+1] != '\n' {
 		return "", fmt.Errorf("protocol error")
 	}
+	if remainingBytes != nil {
+		*remainingBytes -= length
+	}
 	return string(buf[:length]), nil
 }
 
 func readRESPLine(reader *bufio.Reader) (string, error) {
-	line, errRead := reader.ReadString('\n')
-	if errRead != nil {
+	var buf []byte
+	for {
+		fragment, errRead := reader.ReadSlice('\n')
+		if len(buf)+len(fragment) > maxRESPLineLen+2 {
+			return "", fmt.Errorf("protocol error")
+		}
+		if errRead == nil {
+			var line string
+			if len(buf) == 0 {
+				line = string(fragment)
+			} else {
+				buf = append(buf, fragment...)
+				line = string(buf)
+			}
+			line = strings.TrimSuffix(line, "\n")
+			line = strings.TrimSuffix(line, "\r")
+			if len(line) > maxRESPLineLen {
+				return "", fmt.Errorf("protocol error")
+			}
+			return line, nil
+		}
+		if errors.Is(errRead, bufio.ErrBufferFull) {
+			buf = append(buf, fragment...)
+			continue
+		}
 		return "", errRead
 	}
-	line = strings.TrimSuffix(line, "\n")
-	line = strings.TrimSuffix(line, "\r")
-	return line, nil
 }
 
 func writeRedisSimpleString(writer *bufio.Writer, value string) error {