diff --git a/docs/release-control/v6/internal/subsystems/relay-runtime.md b/docs/release-control/v6/internal/subsystems/relay-runtime.md index 5a8445d100..412c74a933 100644 --- a/docs/release-control/v6/internal/subsystems/relay-runtime.md +++ b/docs/release-control/v6/internal/subsystems/relay-runtime.md @@ -86,7 +86,11 @@ Pulse API request so trivial spacing drift does not cause a valid proxied operation to fail at the desktop boundary. The same normalization rule applies to allowlisted proxy header names, so safe headers like `Content-Type` are not silently stripped just because the upstream frame wrapped them in outer -whitespace. +whitespace. That same proxy boundary also owns per-channel abuse control: one +relay channel may not consume an unbounded share of the local HTTP proxy, so +desktop relay runtime must enforce a channel-local request budget and fail +excess DATA frames with owned proxy backpressure instead of letting one +channel starve the rest of the connection. Persisted relay config loading is part of the same owned surface. A missing `relay.enc` file must fall back cleanly to the canonical disabled default so a fresh v6 install or a partially migrated instance does not fail closed just diff --git a/internal/relay/client.go b/internal/relay/client.go index 9277046e25..f83c0d6ab4 100644 --- a/internal/relay/client.go +++ b/internal/relay/client.go @@ -43,11 +43,17 @@ const ( // This prevents unbounded goroutine growth if the relay floods DATA frames. var maxConcurrentDataHandlers = 64 +// relayMaxRequestsPerChannelWindow limits proxied HTTP requests accepted from a +// single relay channel within relayChannelRateLimitWindow. +var relayMaxRequestsPerChannelWindow = 300 + const ( - wsMaxMessageSize = wsReadLimit - wsPongWait = 60 * time.Second - proxyStreamTimeout = 15 * time.Minute - relayOverloadedReason = "relay proxy overloaded" + wsMaxMessageSize = wsReadLimit + wsPongWait = 60 * time.Second + proxyStreamTimeout = 15 * time.Minute + relayOverloadedReason = "relay proxy overloaded" + relayChannelRateLimitWindow = time.Minute + relayRateLimitedReason = "relay proxy rate limited" ) // TokenValidator validates an API token and returns the raw token if valid. @@ -55,9 +61,60 @@ type TokenValidator func(token string) bool // channelState holds per-channel state including auth and encryption. type channelState struct { - apiToken string - encryption *ChannelEncryption // nil until key exchange completes - ephemeral *ecdh.PrivateKey // ephemeral keypair, cleared after handshake + apiToken string + encryption *ChannelEncryption // nil until key exchange completes + ephemeral *ecdh.PrivateKey // ephemeral keypair, cleared after handshake + requestLimiter *relayChannelRequestLimiter +} + +type relayChannelRequestLimiter struct { + mu sync.Mutex + attempts []time.Time + limit int + window time.Duration +} + +func newRelayChannelState(apiToken string) *channelState { + return &channelState{ + apiToken: apiToken, + requestLimiter: newRelayChannelRequestLimiter(relayMaxRequestsPerChannelWindow, relayChannelRateLimitWindow), + } +} + +func newRelayChannelRequestLimiter(limit int, window time.Duration) *relayChannelRequestLimiter { + if limit <= 0 || window <= 0 { + return nil + } + return &relayChannelRequestLimiter{ + limit: limit, + window: window, + } +} + +func (l *relayChannelRequestLimiter) Allow(now time.Time) bool { + if l == nil { + return true + } + + l.mu.Lock() + defer l.mu.Unlock() + + cutoff := now.Add(-l.window) + keep := 0 + for _, attempt := range l.attempts { + if attempt.After(cutoff) { + l.attempts[keep] = attempt + keep++ + } + } + l.attempts = l.attempts[:keep] + + if len(l.attempts) >= l.limit { + return false + } + + l.attempts = append(l.attempts, now) + return true } // ClientDeps holds injectable dependencies for the relay client. @@ -691,7 +748,7 @@ func (c *Client) handleChannelOpen(frame Frame, sendCh chan<- []byte) { // Accept: store channel and echo CHANNEL_OPEN back c.mu.Lock() - c.channels[payload.ChannelID] = &channelState{apiToken: payload.AuthToken} + c.channels[payload.ChannelID] = newRelayChannelState(payload.AuthToken) c.mu.Unlock() c.logger.Info().Uint32("channel", payload.ChannelID).Msg("channel opened") @@ -714,9 +771,11 @@ func (c *Client) handleData(connCtx context.Context, frame Frame, sendCh chan<- state, ok := c.channels[channelID] var enc *ChannelEncryption var apiToken string + var requestLimiter *relayChannelRequestLimiter if ok { enc = state.encryption apiToken = state.apiToken + requestLimiter = state.requestLimiter } c.mu.RUnlock() @@ -738,6 +797,11 @@ func (c *Client) handleData(connCtx context.Context, frame Frame, sendCh chan<- payload = decrypted } + if requestLimiter != nil && !requestLimiter.Allow(time.Now()) { + c.handleRateLimitedData(channelID, payload, enc, sendCh) + return + } + select { case dataLimiter <- struct{}{}: default: @@ -792,6 +856,27 @@ func (c *Client) handleOverloadedData(channelID uint32, payload []byte, enc *Cha queueFrame(sendCh, NewFrame(FrameData, channelID, respPayload), c.logger) } +func (c *Client) handleRateLimitedData(channelID uint32, payload []byte, enc *ChannelEncryption, sendCh chan<- []byte) { + requestID := extractProxyRequestID(payload) + respPayload := c.proxy.errorResponse(requestID, http.StatusTooManyRequests, relayRateLimitedReason) + if enc != nil { + encrypted, err := enc.Encrypt(respPayload) + if err != nil { + c.logger.Warn().Err(err).Uint32("channel", channelID).Msg("Failed to encrypt rate-limit response") + return + } + respPayload = encrypted + } + + c.logger.Warn(). + Uint32("channel", channelID). + Str("request_id", requestID). + Int("limit", relayMaxRequestsPerChannelWindow). + Dur("window", relayChannelRateLimitWindow). + Msg("Relay DATA request rate limit reached for channel") + queueFrame(sendCh, NewFrame(FrameData, channelID, respPayload), c.logger) +} + func extractProxyRequestID(payload []byte) string { var req ProxyRequest if err := json.Unmarshal(payload, &req); err != nil { diff --git a/internal/relay/client_test.go b/internal/relay/client_test.go index 44b4544e03..44ed094f4b 100644 --- a/internal/relay/client_test.go +++ b/internal/relay/client_test.go @@ -1741,6 +1741,158 @@ func TestClient_OverloadedDataReturnsBusyResponse(t *testing.T) { <-errCh } +func TestClient_ChannelRateLimitReturnsTooManyRequests(t *testing.T) { + logger := zerolog.New(zerolog.NewTestWriter(t)) + + origLimit := relayMaxRequestsPerChannelWindow + relayMaxRequestsPerChannelWindow = 1 + defer func() { relayMaxRequestsPerChannelWindow = origLimit }() + + mockAPI := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]string{"path": r.URL.Path}) + })) + defer mockAPI.Close() + localAddr := strings.TrimPrefix(mockAPI.URL, "http://") + + rateLimitedRespCh := make(chan ProxyResponse, 1) + + relayServer := mockRelayServer(t, func(conn *websocket.Conn) { + _, msg, _ := conn.ReadMessage() + frame, _ := DecodeFrame(msg) + if frame.Type != FrameRegister { + return + } + ack, _ := NewControlFrame(FrameRegisterAck, 0, RegisterAckPayload{ + InstanceID: "inst_rate_limit", + SessionToken: "sess_rate_limit", + ExpiresAt: time.Now().Add(time.Hour).Unix(), + }) + ackBytes, _ := EncodeFrame(ack) + _ = conn.WriteMessage(websocket.BinaryMessage, ackBytes) + + time.Sleep(50 * time.Millisecond) + chOpen, _ := NewControlFrame(FrameChannelOpen, 1, ChannelOpenPayload{ + ChannelID: 1, + AuthToken: "valid-token", + }) + chOpenBytes, _ := EncodeFrame(chOpen) + _ = conn.WriteMessage(websocket.BinaryMessage, chOpenBytes) + + _, msg, err := conn.ReadMessage() + if err != nil { + return + } + frame, _ = DecodeFrame(msg) + if frame.Type != FrameChannelOpen { + return + } + + firstReqBytes, _ := json.Marshal(ProxyRequest{ + ID: "req_allowed", + Method: "GET", + Path: "/api/allowed", + }) + firstFrame := NewFrame(FrameData, 1, firstReqBytes) + firstData, _ := EncodeFrame(firstFrame) + _ = conn.WriteMessage(websocket.BinaryMessage, firstData) + + _, msg, err = conn.ReadMessage() + if err != nil { + return + } + frame, _ = DecodeFrame(msg) + if frame.Type != FrameData { + return + } + var firstResp ProxyResponse + if err := json.Unmarshal(frame.Payload, &firstResp); err != nil { + return + } + if firstResp.ID != "req_allowed" || firstResp.Status != http.StatusOK { + return + } + + secondReqBytes, _ := json.Marshal(ProxyRequest{ + ID: "req_limited", + Method: "GET", + Path: "/api/limited", + }) + secondFrame := NewFrame(FrameData, 1, secondReqBytes) + secondData, _ := EncodeFrame(secondFrame) + _ = conn.WriteMessage(websocket.BinaryMessage, secondData) + + _ = conn.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, msg, err = conn.ReadMessage() + if err != nil { + return + } + frame, _ = DecodeFrame(msg) + if frame.Type != FrameData { + return + } + var resp ProxyResponse + if err := json.Unmarshal(frame.Payload, &resp); err != nil { + return + } + rateLimitedRespCh <- resp + }) + defer relayServer.Close() + + cfg := Config{ + Enabled: true, + ServerURL: wsURL(relayServer), + } + + deps := ClientDeps{ + LicenseTokenFunc: func() string { return "test-jwt" }, + TokenValidator: func(token string) bool { return token == "valid-token" }, + LocalAddr: localAddr, + ServerVersion: "1.0.0", + IdentityPubKey: "test-pub-key", + } + + client := NewClient(cfg, deps, logger) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- client.Run(ctx) }() + + select { + case resp := <-rateLimitedRespCh: + if resp.ID != "req_limited" { + t.Errorf("response ID: got %q, want %q", resp.ID, "req_limited") + } + if resp.Status != http.StatusTooManyRequests { + t.Errorf("response status: got %d, want %d", resp.Status, http.StatusTooManyRequests) + } + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for rate-limited response") + } + + cancel() + <-errCh +} + +func TestRelayChannelRequestLimiterPrunesExpiredAttempts(t *testing.T) { + limiter := newRelayChannelRequestLimiter(2, time.Minute) + now := time.Now().UTC() + + if !limiter.Allow(now.Add(-2 * time.Minute)) { + t.Fatal("expected first request to be allowed") + } + if !limiter.Allow(now) { + t.Fatal("expected second request to be allowed") + } + if !limiter.Allow(now.Add(30 * time.Second)) { + t.Fatal("expected expired attempt to be pruned before third request") + } + if limiter.Allow(now.Add(31 * time.Second)) { + t.Fatal("expected limiter to reject once the current window is full") + } +} + func TestNextConsecutiveFailures(t *testing.T) { tests := []struct { name string