Rate limit relay proxy traffic per channel

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

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"
+	relayChannelRateLimitWindow = time.Minute
+	relayRateLimitedReason      = "relay proxy rate limited"
 )
 
 // TokenValidator validates an API token and returns the raw token if valid.
@@ -58,6 +64,57 @@ type channelState struct {
 	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 {

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