console: Add events API rate limiting

config/messages.json

@@ -3545,6 +3545,24 @@
       "file": "subscriptions.go"
     }
   },
+  "error:pkg/console/internal/events:rate_exceeded": {
+    "translations": {
+      "en": "request rate exceeded"
+    },
+    "description": {
+      "package": "pkg/console/internal/events",
+      "file": "ratelimit.go"
+    }
+  },
+  "error:pkg/console/internal/events:unknown_caller": {
+    "translations": {
+      "en": "unknown caller type `{type}`"
+    },
+    "description": {
+      "package": "pkg/console/internal/events",
+      "file": "ratelimit.go"
+    }
+  },
   "error:pkg/crypto/cryptoservices:no_app_key": {
     "translations": {
       "en": "no AppKey specified"

pkg/console/internal/events/events.go

@@ -78,6 +78,12 @@ func (h *eventsHandler) handleEvents(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	rateLimit, err := makeRateLimiter(ctx, h.component.RateLimiter())
+	if err != nil {
+		webhandlers.Error(w, r, err)
+		return
+	}
+
 	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
 		Subprotocols:       []string{protocolV1},
 		InsecureSkipVerify: true, // CORS is not enabled for APIs.
@@ -100,7 +106,7 @@ func (h *eventsHandler) handleEvents(w http.ResponseWriter, r *http.Request) {
 	})
 	for name, f := range map[string]func(context.Context) error{
 		"console_events_mux":   makeMuxTask(m, cancel),
-		"console_events_read":  makeReadTask(conn, m, cancel),
+		"console_events_read":  makeReadTask(conn, m, rateLimit, cancel),
 		"console_events_write": makeWriteTask(conn, m, cancel),
 	} {
 		wg.Add(1)

pkg/console/internal/events/ratelimit.go

@@ -0,0 +1,56 @@
+// Copyright © 2023 The Things Network Foundation, The Things Industries B.V.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package events
+
+import (
+	"context"
+	"fmt"
+
+	"go.thethings.network/lorawan-stack/v3/pkg/auth/rights"
+	"go.thethings.network/lorawan-stack/v3/pkg/errors"
+	"go.thethings.network/lorawan-stack/v3/pkg/ratelimit"
+	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
+)
+
+var (
+	errUnknownCaller = errors.DefineInternal("unknown_caller", "unknown caller type `{type}`")
+	errRateExceeded  = errors.DefineResourceExhausted("rate_exceeded", "request rate exceeded")
+)
+
+func makeRateLimiter(ctx context.Context, limiter ratelimit.Interface) (func() error, error) {
+	authInfo, err := rights.AuthInfo(ctx)
+	if err != nil {
+		return nil, err
+	}
+	resourceID := ""
+	switch method := authInfo.AccessMethod.(type) {
+	case *ttnpb.AuthInfoResponse_ApiKey:
+		resourceID = fmt.Sprintf("api-key:%s", method.ApiKey.ApiKey.Id)
+	case *ttnpb.AuthInfoResponse_OauthAccessToken:
+		resourceID = fmt.Sprintf("access-token:%s", method.OauthAccessToken.Id)
+	case *ttnpb.AuthInfoResponse_UserSession:
+		resourceID = fmt.Sprintf("session-id:%s", method.UserSession.SessionId)
+	// NOTE: *ttnpb.AuthInfoResponse_GatewayToken_ is intentionally left out.
+	default:
+		return nil, errUnknownCaller.WithAttributes("type", fmt.Sprintf("%T", authInfo.AccessMethod))
+	}
+	resource := ratelimit.ConsoleEventsRequestResource(resourceID)
+	return func() error {
+		if limit, _ := limiter.RateLimit(resource); limit {
+			return errRateExceeded.New()
+		}
+		return nil
+	}, nil
+}

pkg/console/internal/events/tasks.go

@@ -33,7 +33,9 @@ func makeMuxTask(m eventsmux.Interface, cancel func(error)) func(context.Context
 	}
 }
 
-func makeReadTask(conn *websocket.Conn, m eventsmux.Interface, cancel func(error)) func(context.Context) error {
+func makeReadTask(
+	conn *websocket.Conn, m eventsmux.Interface, rateLimit func() error, cancel func(error),
+) func(context.Context) error {
 	return func(ctx context.Context) (err error) {
 		defer func() { cancel(err) }()
 		defer func() {
@@ -50,6 +52,9 @@ func makeReadTask(conn *websocket.Conn, m eventsmux.Interface, cancel func(error
 			if err := wsjson.Read(ctx, conn, &request); err != nil {
 				return err
 			}
+			if err := rateLimit(); err != nil {
+				return err
+			}
 			select {
 			case <-ctx.Done():
 				return ctx.Err()

pkg/ratelimit/resource.go

@@ -148,6 +148,15 @@ func ApplicationWebhooksDownResource(ctx context.Context, ids *ttnpb.EndDeviceId
 	}
 }
 
+// ConsoleEventsRequestResource represents a request for events from the Console.
+func ConsoleEventsRequestResource(callerID string) Resource {
+	key := fmt.Sprintf("console:internal:events:request:%s", callerID)
+	return &resource{
+		key:     key,
+		classes: []string{"http:console:internal:events"},
+	}
+}
+
 // NewCustomResource returns a new resource. It is used internally by other components.
 func NewCustomResource(key string, classes ...string) Resource {
 	return &resource{key, classes}

pkg/webui/locales/ja.json

@@ -1885,6 +1885,8 @@
   "error:pkg/console/internal/events/subscriptions:already_subscribed": "",
   "error:pkg/console/internal/events/subscriptions:no_identifiers": "",
   "error:pkg/console/internal/events/subscriptions:not_subscribed": "",
+  "error:pkg/console/internal/events:rate_exceeded": "",
+  "error:pkg/console/internal/events:unknown_caller": "",
   "error:pkg/crypto/cryptoservices:no_app_key": "指定されたAppKeyがありません",
   "error:pkg/crypto/cryptoservices:no_dev_eui": "指定されたDevEUIがありません",
   "error:pkg/crypto/cryptoservices:no_join_eui": "指定されたJoinEUIがありません",