villebro commented on code in PR #45404:
URL: https://github.com/apache/superset/pull/45404#discussion_r4418848462
##########
superset-frontend/src/middleware/realtime.ts:
##########
@@ -267,16 +284,110 @@ const teardownSocket = (): void => {
}
};
+interface Ticket {
+ token: string;
+ expiresInMs: number;
+}
+
+/** Fetch a fresh connection ticket for the current session. */
+const fetchTicket = async (): Promise<Ticket> => {
+ const { json } = await SupersetClient.get({ endpoint: TICKET_ENDPOINT });
+ 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;
+};
+
+const sendAuthFrame = (ws: WebSocket, token: string): void => {
+ ws.send(JSON.stringify({ type: 'auth', token, tab_id: getTabId() }));
+};
+
+// Stop connecting after the server refused this session. Closes any live
socket
+// without scheduling a reconnect.
+const enterUnauthorized = (): void => {
+ teardownSocket();
+ emitState('unauthorized');
Review Comment:
Valid. Fixed in 272b4765cd (was b8eaf29592 before the rebase): realtime.ts
exposes getRealtimeState(), and asyncEvent.ts settles a chart request that
registers while the state is unauthorized or unhealthy immediately, instead of
waiting out pollStaleTimeoutMs. Tests cover a waiter registered after each
terminal state.
##########
superset-frontend/src/middleware/realtime.ts:
##########
@@ -267,16 +301,132 @@ const teardownSocket = (): void => {
}
};
+interface Ticket {
+ token: string;
+ expiresInMs: number;
+}
+
+/** Fetch a fresh connection ticket for the current session. */
+const fetchTicket = async (): Promise<Ticket> => {
+ const { json } = await SupersetClient.get({ endpoint: TICKET_ENDPOINT });
Review Comment:
Valid. The ticket request now passes ignoreUnauthorized: true, so a 401/403
ends in the unauthorized state without running the client's unauthorized
handler (no login redirect on a public dashboard). The realtime test asserts
the flag on the request. Fixed in e76a1f9e56.
##########
superset-frontend/src/middleware/realtime.ts:
##########
@@ -267,16 +301,132 @@ const teardownSocket = (): void => {
}
};
+interface Ticket {
+ token: string;
+ expiresInMs: number;
+}
+
+/** Fetch a fresh connection ticket for the current session. */
+const fetchTicket = async (): Promise<Ticket> => {
+ const { json } = await SupersetClient.get({ endpoint: TICKET_ENDPOINT });
+ 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 => {
+ teardownSocket();
+ emitState('unauthorized');
+};
+
+const refreshRetryDelayMs = (attempt: number): number =>
+ Math.min(REFRESH_RETRY_BASE_MS * 2 ** attempt, REFRESH_RETRY_MAX_MS);
+
+// Send a fresh ticket on the open socket `ws`. A transient ticket failure is
+// retried with backoff until the current ticket expires; past it the server
closes the socket
+// with AUTH_CLOSE_CODE and the reconnect path takes over. The server's
+// `realtime.auth` reply schedules the next refresh.
+function refreshTicket(
+ thisGeneration: number,
+ ws: WebSocket,
+ attempt = 0,
+): void {
+ fetchTicket().then(
+ ({ token }) => {
+ if (thisGeneration !== generation || socket !== ws) return;
+ if (ws.readyState !== WS_OPEN) return;
+ try {
+ sendAuthFrame(ws, token);
+ } catch (err) {
+ logging.warn('Failed to send realtime auth refresh', err);
+ }
+ },
+ err => {
+ if (thisGeneration !== generation || socket !== ws) return;
+ if (isUnauthorizedError(err)) {
+ enterUnauthorized();
+ return;
+ }
+ const delay = refreshRetryDelayMs(attempt);
+ if (Date.now() + delay >= ticketExpiresAtMs) return;
+ refreshTimeoutId = window.setTimeout(
+ () => refreshTicket(thisGeneration, ws, attempt + 1),
+ delay,
+ );
+ },
+ );
+}
+
// Declared as a hoisted function so scheduleReconnect (above) can reference it
// for the mutual reconnect recursion without a use-before-define cycle.
-function openSocket(thisGeneration: number, reason: RealtimeOpenReason): void {
+async function openSocket(
+ thisGeneration: number,
+ reason: RealtimeOpenReason,
+): Promise<void> {
if (thisGeneration !== generation) return;
if (!enabled || !url || typeof WebSocket === 'undefined') return;
- const connectUrl = buildConnectUrl(url);
+ attemptSeq += 1;
+ const attempt = attemptSeq;
+ const isCurrent = () =>
+ thisGeneration === generation && attempt === attemptSeq;
emitState('connecting');
+
+ let ticket: Ticket;
+ try {
+ ticket = await fetchTicket();
+ } catch (err) {
+ if (!isCurrent()) return;
+ if (isUnauthorizedError(err)) {
+ enterUnauthorized();
Review Comment:
Valid. setupGuestClient now calls connectRealtime() once the guest client is
configured, which starts a fresh attempt after the module-load one was refused
(and is a no-op while a socket is already open, so guest-token refreshes don't
churn the socket). Tests: the embedded test asserts the reconnect happens after
setupClient, and realtime.test.ts covers connecting again after an unauthorized
ticket. Fixed in e76a1f9e56.
##########
superset-websocket/src/index.ts:
##########
@@ -715,72 +738,243 @@ 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(),
+const sendAuthOk = (ws: WebSocket, 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),
+ ),
+ },
};
+ ws.send(JSON.stringify(message));
Review Comment:
Valid. sendAuthOk now goes through sendToSocket, so auth acks are subject to
maxSocketBufferBytes like every other outbound message and an over-limit socket
is terminated. Test in spec/auth.test.ts (fails without the change). Fixed in
e76a1f9e56.
--
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]
