sadpandajoe commented on code in PR #45404:
URL: https://github.com/apache/superset/pull/45404#discussion_r4450472932
##########
superset-frontend/src/embedded/index.tsx:
##########
@@ -284,6 +285,10 @@ function setupGuestClient(guestToken: string) {
guestTokenHeaderName: bootstrapData.config?.GUEST_TOKEN_HEADER_NAME,
unauthorizedHandler: guestUnauthorizedHandler,
});
+ // The realtime socket first connects at module load, before a guest token
+ // exists, and its ticket request is refused. Reconnect with the guest
+ // credential; a no-op while a socket is already open.
+ connectRealtime();
Review Comment:
This re-read can't turn realtime on for embedded guests. The embedded shell
is rendered for an anonymous user, so `common_bootstrap_payload()` already sets
`WEBSOCKET_ENABLE` to `false` unless the Public role holds `can_read Realtime`,
and `connectRealtime()` reads that same unchanged bootstrap value after the
guest token arrives. With the guest role granted `can_read Realtime` (and
Public not), no ticket is requested and no socket opens. The new test mocks
`connectRealtime`, so it wouldn't catch this. Should this pass a config with
the websocket enabled once the guest token is set, or is realtime for embedded
guests intentionally off?
##########
superset-frontend/src/middleware/realtime.ts:
##########
@@ -267,16 +301,140 @@ const teardownSocket = (): void => {
}
};
+interface Ticket {
+ token: string;
+ expiresInMs: number;
+}
+
+/**
+ * Fetch a fresh connection ticket for the current session. A 401 must not run
+ * the client's unauthorized handler (a login redirect, or the embedded guest
+ * handler): the realtime transport is optional, and the rejection is handled
by
+ * `isUnauthorizedError`.
+ */
+const fetchTicket = async (): Promise<Ticket> => {
+ const { json } = await SupersetClient.get({
+ endpoint: TICKET_ENDPOINT,
+ ignoreUnauthorized: true,
+ });
+ const { token, expires_in: expiresIn } = (json ?? {}) as {
+ token?: unknown;
+ expires_in?: unknown;
+ };
+ if (typeof token !== 'string' || typeof expiresIn !== 'number') {
+ throw new Error('Malformed realtime ticket response');
+ }
+ return { token, expiresInMs: expiresIn * 1000 };
+};
+
+/**
+ * Whether a ticket request failed because the session may not use the
transport
+ * (no session, or no realtime permission). Retrying cannot succeed.
+ */
+const isUnauthorizedError = (err: unknown): boolean => {
+ const status = (err as { status?: unknown } | null)?.status;
+ return status === 401 || status === 403;
+};
+
+// Tab ids carried by `auth` frames the server has not acknowledged yet, oldest
+// first, and the tab id the server last acknowledged. The server answers every
+// frame in order, so comparing the two tells whether an ack rebound the tab.
+let unackedTabIds: string[] = [];
+let ackedTabId: string | undefined;
+
+const sendAuthFrame = (ws: WebSocket, token: string): void => {
+ const tabId = getTabId();
+ ws.send(JSON.stringify({ type: 'auth', token, tab_id: tabId }));
+ unackedTabIds.push(tabId);
+};
+
+const notifyOpen = (reason: RealtimeOpenReason): void => {
+ openListeners.forEach(listener => {
+ try {
+ listener(reason);
+ } catch (err) {
+ logging.warn('Realtime open-listener error', err);
+ }
+ });
+};
+
+// Stop connecting after the server refused this session. Closes any live
socket
+// without scheduling a reconnect.
+const enterUnauthorized = (): void => {
Review Comment:
`unauthorized` is a dead end for the rest of the page's life in the main
app: nothing but a new `connectRealtime()` call leaves it, and the only caller
is the one-time `asyncEvent` init. Repro: log out in another tab, let this idle
tab's ticket refresh get a 401 (socket torn down, no redirect because of
`ignoreUnauthorized`), log back in, then change a filter here. The chart POST
succeeds, but the waiter is rejected immediately with `Realtime connection
unavailable` and polling is off, so async charts fail until a reload. Before
this change the client kept retrying and recovered once the session was valid
again. Could a 401/403 use a long backoff retry (or re-arm on focus or after a
successful API response) instead of staying terminal?
##########
superset-websocket/src/index.ts:
##########
@@ -715,72 +738,245 @@ export const readSocketIdentity = (
};
};
+/** An inbound `auth` frame: `{type: 'auth', token, tab_id?}`. */
+interface AuthFrame {
+ token: string;
+ tabId?: string;
+}
+
/**
- * WebSocket `connection` event handler. `identity` is the one already verified
- * during the HTTP upgrade (see `httpUpgrade`), so each connection's JWT is
- * verified exactly once.
+ * Parse a raw inbound frame into an `auth` frame, or `null` when it is
anything
+ * else (malformed JSON, another message type, or a missing token). An invalid
or
+ * absent `tab_id` is dropped rather than failing the frame, so the socket
falls
+ * back to principal-grain delivery.
*/
-export const wsConnection = (
- ws: WebSocket,
- identity: SocketIdentity,
- tabId?: string,
-) => {
- const { channel } = identity;
-
- // Refuse the connection if a configured connection limit has been reached,
- // before tracking it against the internal registries. The cap is keyed on
the
- // principal channel (which counts all of a principal's tabs); the per-tab
- // channel is a delivery refinement and is not separately limit-checked.
- const limitReason = connectionLimitReason(channel);
- if (limitReason) {
- statsd.increment('ws_connection_rejected');
- logger.warn(`Refusing connection on channel ${channel}: ${limitReason}`);
- ws.close(CONNECTION_LIMIT_CLOSE_CODE, limitReason);
- return;
+const parseAuthFrame = (raw: WebSocket.RawData): AuthFrame | null => {
+ let parsed: unknown;
+ try {
+ parsed = JSON.parse(raw.toString());
+ } catch {
+ return null;
}
+ if (!parsed || typeof parsed !== 'object') return null;
+ const {
+ type,
+ token,
+ tab_id: tabId,
+ } = parsed as {
+ type?: unknown;
+ token?: unknown;
+ tab_id?: unknown;
+ };
+ if (type !== 'auth' || typeof token !== 'string' || token.length === 0) {
+ return null;
+ }
+ return {
+ token,
+ tabId: typeof tabId === 'string' && isValidTabId(tabId) ? tabId :
undefined,
+ };
+};
- // When the browser advertised a tab id at connect, also register the socket
- // under a per-tab channel derived from its authorized principal channel
- // (`<channel>:<tabId>`). The tab id is client-supplied but always prefixed
by
- // the authorized principal channel, so it can never address another
- // principal's sockets. Task-status can then target one tab, while the
- // principal channel still reaches all of a principal's tabs.
- const tabChannel = tabId ? `${channel}:${tabId}` : undefined;
-
- const socketInstance: SocketInstance = {
- ws,
- channel,
- tabChannel,
- identity,
- pongTs: Date.now(),
+// Sent through the bounded send path: a client replaying auth frames without
+// reading replies hits the same buffer cap as any other outbound message.
+const sendAuthOk = (socketId: string, identity: SocketIdentity): void => {
+ const message: OutboundMessage = {
+ topic: AUTH_REPLY_TOPIC,
+ payload: {
+ status: 'ok',
+ expires_in: Math.max(
+ 0,
+ Math.floor((identity.tokenExpiresAtMs - Date.now()) / 1000),
+ ),
+ },
};
+ sendToSocket(socketId, JSON.stringify(message));
+};
+
+/**
+ * Removes a socket id from a channel's socket list, dropping the channel when
it
+ * becomes empty.
+ */
+const untrackFromChannel = (channel: string, socketId: string): void => {
+ const entry = channels[channel];
+ if (!entry) return;
+ entry.sockets = entry.sockets.filter(id => id !== socketId);
+ if (entry.sockets.length === 0) delete channels[channel];
+};
- // Register once (mints a single socket id), then also index that same id
under
- // the per-tab channel - never call trackClient twice (it would mint a second
- // id and a duplicate registry entry).
- const socketId = trackClient(channel, socketInstance);
+/**
+ * Binds an already tracked socket to the per-tab channel derived from its
+ * authorized principal channel (`<channel>:<tabId>`), replacing a previous
+ * binding. The tab id is client-supplied but always prefixed by the authorized
+ * principal channel, so it can never address another principal's sockets.
+ * Task-status can then target one tab, while the principal channel still
reaches
+ * all of a principal's tabs.
+ */
+const bindTabChannel = (
+ socketId: string,
+ socketInstance: SocketInstance,
+ tabId: string | undefined,
+): void => {
+ const tabChannel = tabId ? `${socketInstance.channel}:${tabId}` : undefined;
+ if (tabChannel === socketInstance.tabChannel) return;
+ if (socketInstance.tabChannel) {
+ untrackFromChannel(socketInstance.tabChannel, socketId);
+ }
if (tabChannel) {
(channels[tabChannel] ??= { sockets: [] }).sockets.push(socketId);
}
- logger.debug(
- `socket ${socketId} connected on channel ${channel}` +
- (tabChannel ? ` (tab channel ${tabChannel})` : ''),
- );
+ socketInstance.tabChannel = tabChannel;
+};
+
+/**
+ * WebSocket `connection` event handler. The connection starts `pending_auth`:
+ * it is tracked in no registry and routed no message until its first `auth`
frame
+ * carries a valid ticket, at which point it is registered on its principal
channel
+ * (and tab channel). Later `auth` frames refresh the ticket on the open
socket.
+ */
+export const wsConnection = (ws: WebSocket) => {
+ // Set once the first `auth` frame is accepted; `undefined` while
pending_auth.
+ let socketId: string | undefined;
+
+ // Reserve capacity at upgrade time so unauthenticated sockets cannot bypass
+ // the configured caps while they wait out the auth window.
+ const { maxPendingConnections } = opts;
+ const limitReason = connectionLimitReason('');
+ if (
+ limitReason ||
+ (maxPendingConnections > 0 && pendingConnections >= maxPendingConnections)
+ ) {
+ statsd.increment('ws_connection_rejected');
+ logger.warn(
+ `Refusing connection: ${limitReason ?? 'pre-auth connection limit
reached'}`,
+ );
+ ws.close(CONNECTION_LIMIT_CLOSE_CODE, limitReason ?? 'too many pending');
+ return;
+ }
+ pendingConnections += 1;
+ let pendingReleased = false;
+ const releasePending = () => {
+ if (pendingReleased) return;
+ pendingReleased = true;
+ pendingConnections -= 1;
+ };
+
+ const authTimer = setTimeout(() => {
+ if (socketId === undefined) {
+ statsd.increment('ws_auth_timeout');
+ releasePending();
+ ws.close(AUTH_CLOSE_CODE, 'authentication timeout');
Review Comment:
Releasing the pending reservation and then calling `ws.close()` leaves the
connection open until the peer answers the close frame, or until `ws`'s ~30s
close timeout destroys it. A client that never replies is therefore uncounted
after the first 10s, and the same holds for the limit-rejection `ws.close()`
calls just above and below. With `maxPendingConnections` set to 100, an
anonymous client that opens sockets, ignores the close frame and reconnects
every 10s can hold several hundred sockets against a cap meant to bound them at
100, and legitimate connects get refused with 1013 meanwhile. Would
`ws.terminate()` on these paths (or releasing the slot from the `close` event)
keep the caps meaningful?
--
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]
