Merge pull request #6667 from TheThingsNetwork/feature/ws-auth

Console protocol authentication and 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

@@ -24,6 +24,7 @@ import (
 	"go.thethings.network/lorawan-stack/v3/pkg/auth/rights"
 	"go.thethings.network/lorawan-stack/v3/pkg/config"
 	"go.thethings.network/lorawan-stack/v3/pkg/console/internal/events/eventsmux"
+	"go.thethings.network/lorawan-stack/v3/pkg/console/internal/events/middleware"
 	"go.thethings.network/lorawan-stack/v3/pkg/console/internal/events/subscriptions"
 	"go.thethings.network/lorawan-stack/v3/pkg/events"
 	"go.thethings.network/lorawan-stack/v3/pkg/log"
@@ -36,6 +37,11 @@ import (
 	"nhooyr.io/websocket"
 )
 
+const (
+	authorizationProtocolPrefix = "ttn.lorawan.v3.header.authorization.bearer."
+	protocolV1                  = "ttn.lorawan.v3.console.internal.events.v1"
+)
+
 // Component is the interface of the component to the events API handler.
 type Component interface {
 	task.Starter
@@ -57,6 +63,7 @@ func (h *eventsHandler) RegisterRoutes(server *web.Server) {
 	router.Use(
 		mux.MiddlewareFunc(webmiddleware.Namespace("console/internal/events")),
 		ratelimit.HTTPMiddleware(h.component.RateLimiter(), "http:console:internal:events"),
+		mux.MiddlewareFunc(middleware.ProtocolAuthentication(authorizationProtocolPrefix)),
 		mux.MiddlewareFunc(webmiddleware.Metadata("Authorization")),
 	)
 	router.Path("/").HandlerFunc(h.handleEvents).Methods(http.MethodGet)
@@ -71,7 +78,14 @@ 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.
 		CompressionMode:    websocket.CompressionContextTakeover,
 	})
@@ -92,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/middleware/auth.go

@@ -0,0 +1,78 @@
+// 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 middleware
+
+import (
+	"fmt"
+	"net/http"
+	"net/textproto"
+	"strings"
+
+	"go.thethings.network/lorawan-stack/v3/pkg/auth"
+)
+
+var (
+	protocolHeader      = textproto.CanonicalMIMEHeaderKey("Sec-WebSocket-Protocol")
+	authorizationHeader = textproto.CanonicalMIMEHeaderKey("Authorization")
+	connectionHeader    = textproto.CanonicalMIMEHeaderKey("Connection")
+	upgradeHeader       = textproto.CanonicalMIMEHeaderKey("Upgrade")
+)
+
+func isWebSocketRequest(r *http.Request) bool {
+	h := r.Header
+	return strings.EqualFold(h.Get(connectionHeader), "upgrade") &&
+		strings.EqualFold(h.Get(upgradeHeader), "websocket")
+}
+
+// ProtocolAuthentication returns a middleware that authenticates WebSocket requests using the subprotocol.
+// The subprotocol must be prefixed with the given prefix.
+// The token is extracted from the subprotocol and used to authenticate the request.
+// If the token is valid, the subprotocol is removed from the request.
+// If the token is invalid, the request is not authenticated.
+func ProtocolAuthentication(prefix string) func(http.Handler) http.Handler {
+	prefixLen := len(prefix)
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if !isWebSocketRequest(r) {
+				next.ServeHTTP(w, r)
+				return
+			}
+			if r.Header.Get(authorizationHeader) != "" {
+				next.ServeHTTP(w, r)
+				return
+			}
+			protocols := strings.Split(strings.TrimSpace(r.Header.Get(protocolHeader)), ",")
+			newProtocols := make([]string, 0, len(protocols))
+			token := ""
+			for _, protocol := range protocols {
+				p := strings.TrimSpace(protocol)
+				if len(p) >= prefixLen && strings.EqualFold(prefix, p[:prefixLen]) {
+					token = p[prefixLen:]
+					continue
+				}
+				newProtocols = append(newProtocols, p)
+			}
+			if _, _, _, err := auth.SplitToken(token); err == nil {
+				if len(newProtocols) > 0 {
+					r.Header.Set(protocolHeader, strings.Join(newProtocols, ","))
+				} else {
+					r.Header.Del(protocolHeader)
+				}
+				r.Header.Set(authorizationHeader, fmt.Sprintf("Bearer %s", token))
+			}
+			next.ServeHTTP(w, r)
+		})
+	}
+}

pkg/console/internal/events/middleware/middleware.go

@@ -0,0 +1,16 @@
+// 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 middleware implements the events middleware.
+package middleware

pkg/console/internal/events/protocol/PROTOCOL.md

@@ -46,6 +46,11 @@ The authentication for the internal API is similar to other APIs available in Th
 
 Upon connecting, no authorization will take place - the endpoint only will check that the provided token is valid (i.e. exists and it is not expired).
 
+The [standard WebSocket API](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API) [does not support custom request headers](https://github.com/whatwg/websockets/issues/16). As a result of this limitation, the backend allows the Console to provide the token as a [protocol](https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/WebSocket#parameters). Specifically, given a `Bearer` token `t`, the following protocols should be provided to the `WebSocket` constructor:
+
+- `ttn.lorawan.v3.console.internal.events.v1`
+- `ttn.lorawan.v3.header.authorization.bearer.t`
+
 ### Message Format
 
 Both requests and responses sent over the WebSocket connection are JSON encoded. All messages are JSON objects and are required to contain at least the following two fields:

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がありません",