fix: tighten tunnel and gateway request handling

Author: dontWatchMeCode

.env.example

@@ -1,8 +1,17 @@
 CINDER_GATEWAY_ADDR=:8081
 CINDER_EDGE_ADDR=:8080
 CINDER_BASE_DOMAIN=example.test
+CINDER_ALLOWED_ORIGINS=
+CINDER_MAX_BODY_BYTES=10485760
+CINDER_GATEWAY_CONNECT_RATE=10
+CINDER_GATEWAY_CONNECT_BURST=20
+CINDER_ROUTE_REGISTER_RATE=5
+CINDER_ROUTE_REGISTER_BURST=10
+CINDER_EDGE_REQUEST_RATE=20
+CINDER_EDGE_REQUEST_BURST=40
 CINDER_DEV_API_KEY=dev-secret-token
 CINDER_DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/cinder_link?sslmode=disable
 CINDER_SERVER_URL=ws://127.0.0.1:8081/ws
+CINDER_TLS_INSECURE=false
 CINDER_API_TOKEN=dev-secret-token
 CINDER_ACCOUNT_ID=acc_personal_dev

README.md

@@ -47,6 +47,14 @@ export CINDER_DATABASE_URL='postgres://postgres:postgres@127.0.0.1:5432/cinder_l
 
 `tunneld` automatically runs SQL migrations on startup when `CINDER_DATABASE_URL` is set.
 
+## TLS and production notes
+
+- Use a TLS-terminating reverse proxy in front of `tunneld` for production `wss` traffic.
+- Keep `CINDER_TLS_INSECURE=false` in production; it exists only for local development against self-signed endpoints.
+- Set `CINDER_ALLOWED_ORIGINS` when browser clients need to reach the gateway from a different origin than the request host.
+- `CINDER_MAX_BODY_BYTES` defaults to `10485760` (10 MiB) and caps request/response bodies until chunked streaming lands.
+- Gateway and edge rate limits are configurable with `CINDER_GATEWAY_CONNECT_RATE`, `CINDER_ROUTE_REGISTER_RATE`, and `CINDER_EDGE_REQUEST_RATE` plus matching `*_BURST` env vars.
+
 ## Quick start
 
 Run the hosted service:

go.mod

@@ -50,6 +50,7 @@ require (
 	golang.org/x/sync v0.17.0 // indirect
 	golang.org/x/sys v0.35.0 // indirect
 	golang.org/x/text v0.29.0 // indirect
+	golang.org/x/time v0.15.0 // indirect
 	golang.org/x/tools v0.36.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250707201910-8d1bb00bc6a7 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7 // indirect

go.sum

@@ -155,6 +155,8 @@ golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
 golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
+golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=

internal/app/tunneld/app.go

@@ -73,8 +73,8 @@ func Run(ctx context.Context, args []string) error {
 	hub := httpstream.NewHub()
 	reconnectManager := reconnect.NewManager(cfg.GraceWindow, sessions.NewResumePersistence(sessionRepo), logger.With(slog.String("component", "reconnect")))
 
-	wsServer := gatewayws.NewServer(logger.With(slog.String("component", "gateway")), tokenValidator, cfg.HeartbeatTimeout, sessionRepo, routeRepo, quotaStore, hub, reconnectManager)
-	edgeServer := httpedge.NewServer(logger.With(slog.String("component", "edge")), routeRepo, hub)
+	wsServer := gatewayws.NewServer(logger.With(slog.String("component", "gateway")), tokenValidator, cfg.HeartbeatTimeout, cfg.BaseDomain, cfg.AllowedOrigins, cfg.GatewayConnectRate, cfg.GatewayConnectBurst, cfg.RouteRegisterRate, cfg.RouteRegisterBurst, sessionRepo, routeRepo, quotaStore, hub, reconnectManager)
+	edgeServer := httpedge.NewServer(logger.With(slog.String("component", "edge")), routeRepo, hub, cfg.MaxBodyBytes, cfg.EdgeRequestRate, cfg.EdgeRequestBurst)
 
 	gatewayHTTP := &http.Server{Addr: cfg.GatewayAddr, Handler: muxGateway(wsServer.Handler(), database)}
 	edgeHTTP := &http.Server{Addr: cfg.EdgeAddr, Handler: edgeServer.Handler()}

internal/client/commands/root.go

@@ -58,7 +58,7 @@ func runAgent(ctx context.Context, args []string) error {
 	}
 	return agent.Run(ctx, agent.Config{
 		SocketPath: *socketPath,
-		Client:     session.Client{ServerURL: *serverURL, APIToken: *apiToken, AccountID: *accountID, Hostname: *name},
+		Client:     session.Client{ServerURL: *serverURL, APIToken: *apiToken, AccountID: *accountID, Hostname: *name, TLSInsecure: defaults.TLSInsecure},
 	})
 }
 
@@ -73,7 +73,7 @@ func runConnect(ctx context.Context, args []string) error {
 		return err
 	}
 
-	client := session.Client{ServerURL: *serverURL, APIToken: *apiToken, AccountID: *accountID, Hostname: *name}
+	client := session.Client{ServerURL: *serverURL, APIToken: *apiToken, AccountID: *accountID, Hostname: *name, TLSInsecure: config.LoadClient().TLSInsecure}
 	result, conn, err := client.Connect(ctx)
 	if err != nil {
 		return err
@@ -105,7 +105,7 @@ func runOpen(ctx context.Context, args []string) error {
 		return fmt.Errorf("--subdomain is required")
 	}
 
-	client := session.Client{ServerURL: *serverURL, APIToken: *apiToken, AccountID: *accountID, Hostname: *name}
+	client := session.Client{ServerURL: *serverURL, APIToken: *apiToken, AccountID: *accountID, Hostname: *name, TLSInsecure: config.LoadClient().TLSInsecure}
 	result, conn, err := client.Connect(ctx)
 	if err != nil {
 		return err

internal/client/session/session.go

@@ -9,8 +9,10 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"os"
 	"runtime"
 	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -21,10 +23,11 @@ import (
 )
 
 type Client struct {
-	ServerURL string
-	APIToken  string
-	AccountID string
-	Hostname  string
+	ServerURL   string
+	APIToken    string
+	AccountID   string
+	Hostname    string
+	TLSInsecure bool
 }
 
 type ConnectResult struct {
@@ -59,8 +62,10 @@ type managedStream struct {
 
 var heartbeatInterval = 15 * time.Second
 
+const httpBodyChunkSize = 16 << 10
+
 func (c Client) Connect(ctx context.Context) (ConnectResult, *ws.Conn, error) {
-	conn, _, err := ws.Dial(ctx, c.ServerURL, nil)
+	conn, _, err := ws.Dial(ctx, c.ServerURL, c.dialOptions())
 	if err != nil {
 		return ConnectResult{}, nil, fmt.Errorf("dial websocket: %w", err)
 	}
@@ -86,7 +91,7 @@ func (c Client) Connect(ctx context.Context) (ConnectResult, *ws.Conn, error) {
 }
 
 func (c Client) Resume(ctx context.Context, sessionID, sessionToken string) (ConnectResult, *ws.Conn, error) {
-	conn, _, err := ws.Dial(ctx, c.ServerURL, nil)
+	conn, _, err := ws.Dial(ctx, c.ServerURL, c.dialOptions())
 	if err != nil {
 		return ConnectResult{}, nil, fmt.Errorf("dial websocket: %w", err)
 	}
@@ -107,6 +112,14 @@ func (c Client) Resume(ctx context.Context, sessionID, sessionToken string) (Con
 	return result, conn, nil
 }
 
+func (c Client) dialOptions() *ws.DialOptions {
+	if !c.TLSInsecure {
+		return nil
+	}
+	transport := &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
+	return &ws.DialOptions{HTTPClient: &http.Client{Transport: transport}}
+}
+
 func (c Client) RegisterRoute(ctx context.Context, conn *ws.Conn, subdomain string, target map[string]interface{}) (protocol.RouteRegisteredPayload, error) {
 	frame := protocol.NewEnvelope(protocol.TypeRouteRegister, protocol.ControlStreamID, protocol.RouteRegisterPayload{
 		RequestedSubdomain: subdomain,
@@ -515,6 +528,9 @@ func proxyRequest(ctx context.Context, httpClient *http.Client, manager *streamM
 	if err != nil {
 		return err
 	}
+	if headers.ContentLength >= 0 {
+		req.ContentLength = headers.ContentLength
+	}
 	if headers.Host != "" {
 		req.Host = headers.Host
 	}
@@ -536,16 +552,9 @@ func proxyRequest(ctx context.Context, httpClient *http.Client, manager *streamM
 	if err := manager.write(ctx, responseHeaders); err != nil {
 		return err
 	}
-
-	respBody, err := io.ReadAll(resp.Body)
-	if err != nil {
+	if err := sendBodyReader(ctx, manager, env, protocol.TypeHTTPRespBody, resp.Body, maxTunnelBodyBytes()); err != nil {
 		return err
 	}
-	if len(respBody) > 0 {
-		if err := manager.write(ctx, protocol.Envelope{Version: 1, Type: protocol.TypeHTTPRespBody, SessionID: env.SessionID, StreamID: env.StreamID, RouteID: env.RouteID, Timestamp: time.Now().UTC(), Payload: protocol.HTTPBodyPayload{Base64: true, Data: base64.StdEncoding.EncodeToString(respBody)}}); err != nil {
-			return err
-		}
-	}
 	return manager.write(ctx, protocol.Envelope{Version: 1, Type: protocol.TypeHTTPRespEnd, SessionID: env.SessionID, StreamID: env.StreamID, RouteID: env.RouteID, Timestamp: time.Now().UTC(), Payload: map[string]any{}})
 }
 
@@ -599,6 +608,66 @@ func decodeBody(payload protocol.HTTPBodyPayload) ([]byte, error) {
 	return base64.StdEncoding.DecodeString(payload.Data)
 }
 
+func maxTunnelBodyBytes() int64 {
+	const fallback = int64(10 << 20)
+	value := strings.TrimSpace(os.Getenv("CINDER_MAX_BODY_BYTES"))
+	if value == "" {
+		return fallback
+	}
+	parsed, err := strconv.ParseInt(value, 10, 64)
+	if err != nil || parsed <= 0 {
+		return fallback
+	}
+	return parsed
+}
+
+func sendBodyReader(ctx context.Context, manager *streamManager, env protocol.IncomingEnvelope, frameType string, body io.Reader, maxBytes int64) error {
+	if body == nil {
+		return nil
+	}
+	remaining := maxBytes
+	buf := make([]byte, httpBodyChunkSize)
+	for {
+		readSize := len(buf)
+		if maxBytes > 0 && int64(readSize) > remaining+1 {
+			readSize = int(remaining + 1)
+		}
+		n, err := body.Read(buf[:readSize])
+		if n > 0 {
+			if maxBytes > 0 && int64(n) > remaining {
+				return fmt.Errorf("body exceeds max size of %d bytes", maxBytes)
+			}
+			if maxBytes > 0 {
+				remaining -= int64(n)
+			}
+			if err := manager.write(ctx, protocol.Envelope{Version: 1, Type: frameType, SessionID: env.SessionID, StreamID: env.StreamID, RouteID: env.RouteID, Timestamp: time.Now().UTC(), Payload: protocol.HTTPBodyPayload{Base64: true, Data: base64.StdEncoding.EncodeToString(buf[:n])}}); err != nil {
+				return err
+			}
+		}
+		if err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return err
+		}
+	}
+}
+
+func readBodyWithLimit(r io.Reader, maxBytes int64) ([]byte, error) {
+	if maxBytes <= 0 {
+		return io.ReadAll(r)
+	}
+	limited := io.LimitReader(r, maxBytes+1)
+	body, err := io.ReadAll(limited)
+	if err != nil {
+		return nil, err
+	}
+	if int64(len(body)) > maxBytes {
+		return nil, fmt.Errorf("body exceeds max size of %d bytes", maxBytes)
+	}
+	return body, nil
+}
+
 func stripHopByHopHeaders(headers map[string][]string) map[string][]string {
 	if len(headers) == 0 {
 		return nil

internal/client/session/session_test.go

@@ -8,6 +8,8 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"os"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -55,6 +57,17 @@ func TestClosePayloadFromErrorDefaults(t *testing.T) {
 	}
 }
 
+func TestClientDialOptionsUsesInsecureTLSWhenRequested(t *testing.T) {
+	options := (Client{TLSInsecure: true}).dialOptions()
+	if options == nil || options.HTTPClient == nil {
+		t.Fatal("expected dial options with HTTP client")
+	}
+	transport, ok := options.HTTPClient.Transport.(*http.Transport)
+	if !ok || transport.TLSClientConfig == nil || !transport.TLSClientConfig.InsecureSkipVerify {
+		t.Fatal("expected insecure TLS transport when TLSInsecure is enabled")
+	}
+}
+
 func TestReadConnectResultAuthOK(t *testing.T) {
 	conn := newTestWebsocketConn(t, protocol.NewEnvelope(protocol.TypeAuthOK, protocol.ControlStreamID, protocol.AuthOKPayload{
 		SessionID:          "ses_123",
@@ -382,6 +395,109 @@ func TestManagedStreamSendDoesNotDeadlockOnClose(t *testing.T) {
 	}
 }
 
+func TestReadRequestBodyConcatenatesChunks(t *testing.T) {
+	stream := make(chan protocol.IncomingEnvelope, 3)
+	stream <- protocol.IncomingEnvelope{Type: protocol.TypeHTTPReqBody, Payload: []byte(`{"base64":true,"data":"aGVs"}`)}
+	stream <- protocol.IncomingEnvelope{Type: protocol.TypeHTTPReqBody, Payload: []byte(`{"base64":true,"data":"bG8="}`)}
+	stream <- protocol.IncomingEnvelope{Type: protocol.TypeHTTPReqEnd, Payload: []byte(`{}`)}
+	close(stream)
+
+	data, err := readRequestBody(context.Background(), stream)
+	if err != nil {
+		t.Fatalf("read request body: %v", err)
+	}
+	if string(data) != "hello" {
+		t.Fatalf("expected concatenated body hello, got %q", string(data))
+	}
+}
+
+func TestSendBodyReaderSplitsLargePayloadIntoMultipleFrames(t *testing.T) {
+	serverConn := make(chan *ws.Conn, 1)
+	frameCh := make(chan protocol.IncomingEnvelope, 4)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := ws.Accept(w, r, nil)
+		if err != nil {
+			t.Errorf("accept websocket: %v", err)
+			return
+		}
+		serverConn <- conn
+		for i := 0; i < 4; i++ {
+			_, data, err := conn.Read(context.Background())
+			if err != nil {
+				return
+			}
+			var env protocol.IncomingEnvelope
+			if err := json.Unmarshal(data, &env); err != nil {
+				t.Errorf("decode frame: %v", err)
+				return
+			}
+			frameCh <- env
+		}
+	}))
+	defer server.Close()
+
+	clientConn, _, err := ws.Dial(context.Background(), "ws"+server.URL[len("http"):], nil)
+	if err != nil {
+		t.Fatalf("dial test websocket: %v", err)
+	}
+	defer clientConn.CloseNow()
+	conn := <-serverConn
+	defer conn.CloseNow()
+
+	manager := &streamManager{conn: clientConn, streams: make(map[string]*managedStream), closed: make(map[string]struct{}), newStreams: make(chan protocol.IncomingEnvelope, 1)}
+	payload := strings.Repeat("a", httpBodyChunkSize*2+17)
+	env := protocol.IncomingEnvelope{SessionID: "ses_1", StreamID: "str_1", RouteID: "rte_1"}
+	if err := sendBodyReader(context.Background(), manager, env, protocol.TypeHTTPRespBody, strings.NewReader(payload), int64(len(payload))); err != nil {
+		t.Fatalf("send body reader: %v", err)
+	}
+
+	total := 0
+	frames := 0
+	for i := 0; i < 3; i++ {
+		got := <-frameCh
+		if got.Type != protocol.TypeHTTPRespBody {
+			t.Fatalf("expected body frame, got %s", got.Type)
+		}
+		var bodyPayload protocol.HTTPBodyPayload
+		if err := got.UnmarshalPayload(&bodyPayload); err != nil {
+			t.Fatalf("unmarshal body payload: %v", err)
+		}
+		chunk, err := decodeBody(bodyPayload)
+		if err != nil {
+			t.Fatalf("decode chunk: %v", err)
+		}
+		total += len(chunk)
+		frames++
+	}
+	if frames != 3 || total != len(payload) {
+		t.Fatalf("expected 3 frames totaling %d bytes, got %d frames totaling %d bytes", len(payload), frames, total)
+	}
+}
+
+func TestReadBodyWithLimitRejectsOversizedBody(t *testing.T) {
+	body, err := readBodyWithLimit(strings.NewReader("12345"), 4)
+	if err == nil {
+		t.Fatalf("expected limit error, got body %q", string(body))
+	}
+}
+
+func TestMaxTunnelBodyBytesUsesEnvOverride(t *testing.T) {
+	oldValue, hadValue := os.LookupEnv("CINDER_MAX_BODY_BYTES")
+	t.Cleanup(func() {
+		if hadValue {
+			_ = os.Setenv("CINDER_MAX_BODY_BYTES", oldValue)
+		} else {
+			_ = os.Unsetenv("CINDER_MAX_BODY_BYTES")
+		}
+	})
+	if err := os.Setenv("CINDER_MAX_BODY_BYTES", "2048"); err != nil {
+		t.Fatalf("set env: %v", err)
+	}
+	if got := maxTunnelBodyBytes(); got != 2048 {
+		t.Fatalf("expected env override to be used, got %d", got)
+	}
+}
+
 func newTestWebsocketConn(t *testing.T, env protocol.Envelope) *ws.Conn {
 	t.Helper()
 	serverConn := make(chan *ws.Conn, 1)