Added streaming for httpmuxer (antoniomika#255)

* Added streaming for httpmuxer

* Fix gzip response checking

README.md

@@ -341,7 +341,7 @@ Flags:
       --cleanup-unbound-timeout duration                        Duration to wait before cleaning up an unbound (unforwarded) connection (default 5s)
   -c, --config string                                           Config file (default "config.yml")
       --debug                                                   Enable debugging information
-      --debug-interval duration                                 The duration to wait between each debug loop output if debug is true (default 2s)
+      --debug-interval duration                                 Duration to wait between each debug loop output if debug is true (default 2s)
   -d, --domain string                                           The root domain for HTTP(S) multiplexing that will be appended to subdomains (default "ssi.sh")
       --force-requested-aliases                                 Force the aliases used to be the one that is requested. Will fail the bind if it exists already
       --force-requested-ports                                   Force the ports used to be the one that is requested. Will fail the bind if it exists already
@@ -392,6 +392,7 @@ Flags:
                                                                 to instead of responding with a 404 (default "https://github.com/antoniomika/sish")
       --rewrite-host-header                                     Force rewrite the host header if the user provides host-header=host.com (default true)
       --service-console                                         Enable the service console for each service and send the info to connected clients
+      --service-console-max-content-length int                  The max content length before we stop reading the response body (default -1)
   -m, --service-console-token string                            The token to use for service console access. Auto generated if empty for each connected tunnel
       --sni-load-balancer                                       Enable the SNI load balancer (multiple clients can bind the same SNI domain/port)
       --sni-proxy                                               Enable the use of SNI proxying

cmd/sish.go

@@ -130,6 +130,7 @@ func init() {
 	rootCmd.PersistentFlags().IntP("log-to-file-max-size", "", 500, "The maximum size of outputed log files in megabytes")
 	rootCmd.PersistentFlags().IntP("log-to-file-max-backups", "", 3, "The maxium number of rotated logs files to keep")
 	rootCmd.PersistentFlags().IntP("log-to-file-max-age", "", 28, "The maxium number of days to store log output in a file")
+	rootCmd.PersistentFlags().IntP("service-console-max-content-length", "", -1, "The max content length before we stop reading the response body")
 
 	rootCmd.PersistentFlags().DurationP("debug-interval", "", 2*time.Second, "Duration to wait between each debug loop output if debug is true")
 	rootCmd.PersistentFlags().DurationP("idle-connection-timeout", "", 5*time.Second, "Duration to wait for activity before closing a connection for all reads and writes")

httpmuxer/httpmuxer.go

@@ -6,6 +6,7 @@ package httpmuxer
 import (
 	"bytes"
 	"encoding/base64"
+	"encoding/json"
 	"fmt"
 	"io"
 	"log"
@@ -121,6 +122,28 @@ func Start(state *utils.State) {
 			}
 		}
 
+		routeInfo, routeOk := param.Keys["broadcastRoute"].(string)
+		routeData, dataOk := param.Keys["broadcastData"].(map[string]any)
+
+		if routeOk && dataOk {
+			roundTime := 10 * time.Microsecond
+			if param.Latency > time.Second {
+				roundTime = 10 * time.Millisecond
+			}
+
+			routeData["currentTime"] = param.TimeStamp.Format(viper.GetString("time-format"))
+			routeData["requestTime"] = param.Latency.Round(roundTime).String()
+			routeData["responseCode"] = param.StatusCode
+			routeData["responseStatus"] = fmt.Sprintf("%d %s", param.StatusCode, http.StatusText(param.StatusCode))
+
+			jsonData, err := json.Marshal(routeData)
+			if err != nil {
+				log.Println("unable to marshal json data", err)
+			} else {
+				state.Console.BroadcastRoute(routeInfo, jsonData)
+			}
+		}
+
 		return logLine
 	}), gin.Recovery(), func(c *gin.Context) {
 		c.Set("originalURI", c.Request.RequestURI)
@@ -263,13 +286,22 @@ func Start(state *utils.State) {
 			return
 		}
 
-		reqBody, err := io.ReadAll(c.Request.Body)
-		if err != nil {
-			log.Println("Error reading request body:", err)
-			return
+		var err error
+		var reqBody []byte
+
+		if viper.GetInt64("service-console-max-content-length") == -1 || (viper.GetInt64("service-console-max-content-length") > -1 && c.Request.ContentLength > -1 && c.Request.ContentLength < viper.GetInt64("service-console-max-content-length")) {
+			reqBody, err = io.ReadAll(c.Request.Body)
+			if err != nil {
+				log.Println("Error reading request body:", err)
+				return
+			}
 		}
 
-		c.Request.Body = io.NopCloser(bytes.NewBuffer(reqBody))
+		if reqBody != nil {
+			c.Request.Body = io.NopCloser(bytes.NewBuffer(reqBody))
+		} else {
+			reqBody = []byte("{\"_sish_status\": false, \"_sish_message\": \"request body size exceeds limit for service console\"}")
+		}
 
 		err = forward.ResponseModifier(ResponseModifier(state, hostname, reqBody, c, currentListener))(currentListener.Forward)
 		if err != nil {

httpmuxer/proxy.go

@@ -5,13 +5,11 @@ import (
 	"compress/gzip"
 	"crypto/tls"
 	"encoding/base64"
-	"encoding/json"
 	"io"
 	"log"
 	"net"
 	"net/http"
 	"strings"
-	"time"
 
 	"github.com/antoniomika/sish/utils"
 	"github.com/gin-gonic/gin"
@@ -46,57 +44,51 @@ func RoundTripper() *http.Transport {
 func ResponseModifier(state *utils.State, hostname string, reqBody []byte, c *gin.Context, currentListener *utils.HTTPHolder) func(*http.Response) error {
 	return func(response *http.Response) error {
 		if viper.GetBool("admin-console") || viper.GetBool("service-console") {
-			resBody, err := io.ReadAll(response.Body)
-			if err != nil {
-				log.Println("Error reading response for webconsole:", err)
-			}
-
-			response.Body = io.NopCloser(bytes.NewBuffer(resBody))
-
-			startTime := c.GetTime("startTime")
-			currentTime := time.Now()
-			diffTime := currentTime.Sub(startTime)
+			var err error
+			var resBody []byte
 
-			roundTime := 10 * time.Microsecond
-			if diffTime > time.Second {
-				roundTime = 10 * time.Millisecond
-			}
-
-			if response.Header.Get("Content-Encoding") == "gzip" {
-				gzData := bytes.NewBuffer(resBody)
-				gzReader, err := gzip.NewReader(gzData)
+			if viper.GetInt64("service-console-max-content-length") == -1 || (viper.GetInt64("service-console-max-content-length") > -1 && response.ContentLength > -1 && response.ContentLength < viper.GetInt64("service-console-max-content-length")) {
+				resBody, err = io.ReadAll(response.Body)
 				if err != nil {
-					log.Println("Error reading gzip data:", err)
+					log.Println("Error reading response body:", err)
 				}
+			}
 
-				resBody, err = io.ReadAll(gzReader)
-				if err != nil {
-					log.Println("Error reading gzip data:", err)
+			if resBody != nil {
+				response.Body = io.NopCloser(bytes.NewBuffer(resBody))
+
+				if response.Header.Get("Content-Encoding") == "gzip" {
+					gzData := bytes.NewBuffer(resBody)
+					gzReader, err := gzip.NewReader(gzData)
+					if err != nil {
+						log.Println("Error reading gzip data:", err)
+					}
+
+					resBody, err = io.ReadAll(gzReader)
+					if err != nil {
+						log.Println("Error reading gzip data:", err)
+					}
 				}
+			} else {
+				resBody = []byte("{\"_sish_status\": false, \"_sish_message\": \"response body size exceeds limit for service console\"}")
 			}
 
+			startTime := c.GetTime("startTime")
+
 			requestHeaders := c.Request.Header.Clone()
 			requestHeaders.Add("Host", hostname)
 
-			data, err := json.Marshal(map[string]any{
+			data := map[string]any{
 				"startTime":          startTime,
 				"startTimePretty":    startTime.Format(viper.GetString("time-format")),
-				"currentTime":        currentTime,
 				"requestIP":          c.ClientIP(),
-				"requestTime":        diffTime.Round(roundTime).String(),
 				"requestMethod":      c.Request.Method,
 				"requestUrl":         c.Request.URL,
 				"originalRequestURI": c.GetString("originalURI"),
 				"requestHeaders":     requestHeaders,
 				"requestBody":        base64.StdEncoding.EncodeToString(reqBody),
 				"responseHeaders":    response.Header,
-				"responseCode":       response.StatusCode,
-				"responseStatus":     response.Status,
 				"responseBody":       base64.StdEncoding.EncodeToString(resBody),
-			})
-
-			if err != nil {
-				log.Println("Error marshaling json for webconsole:", err)
 			}
 
 			if response.Request != nil {
@@ -108,7 +100,8 @@ func ResponseModifier(state *utils.State, hostname string, reqBody []byte, c *gi
 				c.Set("proxySocket", string(hostLocation))
 			}
 
-			state.Console.BroadcastRoute(currentListener.HTTPUrl.String(), data)
+			c.Set("broadcastRoute", currentListener.HTTPUrl.String())
+			c.Set("broadcastData", data)
 		}
 
 		return nil

sshmuxer/httphandler.go

@@ -34,6 +34,7 @@ func handleHTTPListener(check *channelForwardMsg, stringPort string, requestMess
 		rT := httpmuxer.RoundTripper()
 
 		fwd, err := forward.New(
+			forward.Stream(true),
 			forward.PassHostHeader(true),
 			forward.RoundTripper(rT),
 			forward.WebsocketRoundTripper(rT),