sadpandajoe commented on code in PR #45404: URL: https://github.com/apache/superset/pull/45404#discussion_r4419678171
########## superset-websocket/spec/auth.test.ts: ########## @@ -0,0 +1,354 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + +/** + * End-to-end tests of the in-band authentication state machine over real + * WebSocket connections on a loopback port: pending_auth -> authenticated, + * in-band refresh, and the close codes that tell a client what to do next. + */ +import jwt from 'jsonwebtoken'; +import * as http from 'http'; +import { duplexPair, type Duplex } from 'stream'; +import { afterAll, beforeAll, beforeEach, expect, test, vi } from 'vitest'; +import { WebSocket } from 'ws'; +import config from '../config.test.json' with { type: 'json' }; +import * as server from '../src/index'; + +vi.mock('ioredis', () => ({ + Redis: vi.fn().mockImplementation(function () { + return { subscribe: vi.fn(), on: vi.fn() }; + }), +})); + +type Frame = { topic: string; payload: Record<string, unknown> }; + +const signTicket = ( + channel = 'user:5', + expiresInSeconds = 3600, + secret = config.jwtSecret, +) => + jwt.sign( + { + channel, + sub: channel.replace('user:', ''), + principal_type: 'user', + aud: 'superset-websocket', + iss: 'superset', + exp: Math.floor(Date.now() / 1000) + expiresInSeconds, + }, + secret, + ); + +// The server under test is driven over in-memory duplex streams rather than a +// TCP listener: the HTTP parser, the upgrade handshake and the WebSocket framing +// are all real, only the transport is a pipe. +let httpServer: http.Server; +const clients: WebSocket[] = []; + +/** Give a stream the socket methods `http` and `ws` call on their transport. */ +const asSocket = (stream: Duplex): Duplex => + Object.assign(stream, { + setTimeout: () => stream, + setNoDelay: () => stream, + setKeepAlive: () => stream, + ref: () => stream, + unref: () => stream, + remoteAddress: '127.0.0.1', + }); + +/** A client socket whose peer is the server under test. */ +const connectToServer = (): Duplex => { + const [clientSide, serverSide] = duplexPair(); + httpServer.emit('connection', asSocket(serverSide)); + return asSocket(clientSide); +}; + +const newClient = ( + options: ConstructorParameters<typeof WebSocket>[2] = {}, +) => { + const ws = new WebSocket('ws://localhost/', { + createConnection: () => connectToServer() as never, + ...options, + }); + clients.push(ws); + return ws; +}; + +beforeAll(() => { + httpServer = http.createServer(); + httpServer.on('upgrade', server.httpUpgrade); +}); + +beforeEach(() => { + server.resetState(); + clients.splice(0).forEach(client => client.terminate()); Review Comment: `resetState()` zeroes `pendingConnections` here, but the previous test's sockets are only `terminate()`d, so their server-side `close` handlers (which call `releasePending()`) run afterwards and decrement the freshly reset counter. After `unauthenticated sockets count toward maxTotalConnections` leaves one pending socket, that late decrement can land inside the next `maxPendingConnections` test, so its second `open()` is admitted instead of getting a 1013 and the test becomes timing-dependent. Could `beforeEach` terminate the clients and await their close before calling `resetState()`? ########## superset/websocket/api.py: ########## @@ -0,0 +1,78 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. +"""REST API issuing connection tickets for the realtime websocket transport.""" + +from __future__ import annotations + +from flask import current_app, Response +from flask_appbuilder.api import expose, protect, safe +from flask_appbuilder.security.decorators import permission_name + +from superset.extensions import event_logger +from superset.views.base_api import BaseSupersetApi, statsd_metrics +from superset.websocket.channel import get_realtime_principal, mint_channel_token +from superset.websocket.permissions import REALTIME_NOTIFICATION_RESOURCE + + +class RealtimeRestApi(BaseSupersetApi): + """Issues short-lived tickets the browser presents to the websocket server.""" + + resource_name = "realtime" + class_permission_name = REALTIME_NOTIFICATION_RESOURCE + allow_browser_login = True + openapi_spec_tag = "Realtime" + + @expose("/token/", methods=("GET",)) + @event_logger.log_this + @protect() + @safe + @statsd_metrics + @permission_name("read") + def token(self) -> Response: + """Get a websocket connection ticket for the current principal. + --- + get: + summary: Get a websocket connection ticket + description: >- + Returns a short-lived JWT the client sends as the first frame of a + websocket connection, and again on the open socket to refresh it + before it expires. + responses: + 200: + description: A connection ticket and its lifetime in seconds + content: + application/json: + schema: + type: object + properties: + token: + type: string + expires_in: + type: integer + 401: + $ref: '#/components/responses/401' + 403: + $ref: '#/components/responses/403' + """ + principal = get_realtime_principal() + if principal is None: + return self.response_401() + return self.response( Review Comment: The response carries a bearer credential but sets no `Cache-Control`. An embedded guest authenticates with `X-GuestToken` and no cookie, so a shared cache in front of Superset that stores GET responses could serve guest A's ticket to guest B, who could then authenticate on A's principal channel until the ticket expires. Should this response be marked `no-store` (and ideally `private`)? ########## SECURITY.md: ########## @@ -90,7 +90,7 @@ Asynchronous execution paths do not create a separate data-access capability. A Task metadata is itself a request-scoped resource. Non-admin users and embedded guests may read or cancel only tasks they are subscribed to or that otherwise represent work they are entitled to observe; Admin may observe and manage tasks as part of the trusted operational boundary. A bug that lets a principal create, read, join, cancel, or receive task state for work outside the role and capability matrix is in scope. -Realtime transports, including WebSocket delivery backed by Redis or Valkey Pub/Sub, are notification mechanisms rather than authorization sources. WebSocket notification access is controlled by `can_read` on the `Realtime` resource. The broadcast scope is authenticated-global, not public: it reaches every authenticated realtime socket, and an anonymous request receives no realtime principal, no JWT cookie, and therefore no socket, so it never receives these messages (true anonymous/Public-role realtime is not offered and would require a separate, restricted model). Broadcast Pub/Sub messages, such as list-view entity-change events, must be context-free nudges; sensitive or authoritative state must not be published on the broadcast scope. Targeted Pub/Sub messages may carry task state only when the producer names routing keys derived from an authorized identity, such as a task subscriber's principal channel (or a per-tab channel derived from it); the producer validates every routing key against the task's own subscriber principals before publishing, and the websocket server forwards the payload only to sockets bound to those keys. Full data and result payloads must still be fetched through the normal protected REST API or cache-read path. Redis Streams used for task completion, dependency, and lock-release signalling are likewise coordination signals; the metastore or cache entry they wake a consumer to read remains the source of truth. +Realtime transports, including WebSocket delivery backed by Redis or Valkey Pub/Sub, are notification mechanisms rather than authorization sources. WebSocket notification access is controlled by `can_read` on the `Realtime` resource. The broadcast scope is authenticated-global, not public: it reaches every authenticated realtime socket, and an anonymous request receives no realtime principal, no connection ticket, and therefore no socket, so it never receives these messages (true anonymous/Public-role realtime is not offered and would require a separate, restricted model). Broadcast Pub/Sub messages, such as list-view entity-change events, must be context-free nudges; sensitive or authoritative state must not be published on the broadcast scope. Targeted Pub/Sub messages may carry task state only when the producer names routing keys derived from an authorized identity, such as a task subscriber's principal channel (or a per-tab channel derived from it); the producer validates every routing key against the task's own subscriber principals before publishing, and the websocket server forwards the payload only to sockets bound to those keys. Full data and result payloads must still be fetched through the normal protected REST API or cache-read path. Redis Streams used for task completion, dependency, and lock-release signalling are likewise coordination signals; the metastore or cache entry they wake a consumer to read remains the source of truth. The realtime notification permission is distinct from the permission to read the underlying object. It controls whether a principal receives push notifications, not whether they may read the object once they call the protected REST API. Existing websocket connections are authorized by the JWT accepted at upgrade time; permission revocation after token minting is bounded by `WEBSOCKET_JWT_EXPIRATION_SECONDS` plus the websocket server's socket-check interval. Redis Streams are internal server-to-server coordination primitives and should not be directly exposed as an end-user subscription surface. Review Comment: This sentence still says existing connections are authorized by the JWT accepted at upgrade time, but the upgrade now carries no credential and the socket is authorized by the `auth` frame and each in-band refresh. An operator reading the policy would take the wrong model for when revocation takes effect. Should it say connections are authorized by the most recent ticket presented on the socket? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
