Fix cross-user identity binding race in PROXY auth mode websocket setup

server.js

@@ -28,6 +28,7 @@ import { getCDAPConfig } from 'server/cdap-config';
 import { applyGraphQLMiddleware } from 'gql/graphql';
 import { getHostName } from 'server/config/hostname';
 import middleware404 from 'server/middleware-404';
+import { createSocketIdentityStore } from 'server/socket-identity';
 
 var cdapConfig,
   securityConfig,
@@ -215,14 +216,38 @@ getCDAPConfig()
      *    requests.
      * 3. The client will not know about the auth token either.
      * 4. Once the client reaches CDAP UI, the proxy would have already authenticated the user.
-     * 5. The request to upgrade websocket connection should already have the auth token and the user id
+     * 5. The request to establish the sockjs session should already have the auth token and the user id
      * 6. We take those values and add to the connection object (sockjs connection object)
      * 7. This then gets picked up at the aggregator module that actually makes the call to the
      *    backend along with these in the request header.
      * 8. Upon receiving the response, we remove these from the request object and send it back
      *    to the client as if no authentication exists.
+     *
+     * Step 5/6 need one correction versus how this used to work: sockjs sessions aren't only
+     * established over a websocket upgrade. Whenever a websocket upgrade isn't available
+     * (some corporate proxies, older browsers), sockjs transparently falls back to plain HTTP
+     * transports (xhr-streaming, xhr-polling, eventsource, ...), which never fire Node's
+     * 'upgrade' event at all. The auth token / user id must therefore be captured per sockjs
+     * session id (parsed out of the request URL) rather than in one shared variable -- a shared
+     * variable is clobbered by whichever request happens to arrive last, silently binding one
+     * user's session to a different, unrelated user's identity (in the fallback-transport case,
+     * every single session gets bound this way, since 'upgrade' never fires for them at all).
      */
-    let authToken, userid;
+    const SOCKJS_PREFIX = '/_sock';
+    const socketIdentities = createSocketIdentityStore({
+      prefix: SOCKJS_PREFIX,
+      cdapConfig,
+      getAuthHeaderFromRawCookies,
+    });
+
+    function isAllowedOrigin(req, allowMissing = false) {
+      const origin = req.headers.origin;
+      if (!origin) {
+        return allowMissing;
+      }
+      return allowedOrigin.indexOf(origin) !== -1;
+    }
+
     sockServer.on('connection', function(c) {
       if (!c) {
         log.error('Connection requested, but no connection available');
@@ -231,8 +256,9 @@ getCDAPConfig()
       log.debug('[SOCKET OPEN] Connection to client "' + c.id + '" opened');
       // @ts-ignore
       var a = new Aggregator(c, { ...cdapConfig, ...securityConfig });
-      c.authToken = authToken;
-      c.userid = userid;
+      const identity = socketIdentities.consume(c.url);
+      c.authToken = identity.authToken;
+      c.userid = identity.userid;
       wsConnections[c.id] = c;
       c.on('close', function() {
         log.debug('Cleaning out aggregator: ' + JSON.stringify(a.connection.id));
@@ -243,20 +269,29 @@ getCDAPConfig()
       });
     });
 
-    sockServer.installHandlers(server, { prefix: '/_sock' });
+    sockServer.installHandlers(server, { prefix: SOCKJS_PREFIX });
     server.addListener('upgrade', function(req, socket) {
-      req.headers.authorization = getAuthHeaderFromRawCookies(req);
-      authToken = req.headers.authorization;
-      const userIdProperty = cdapConfig['security.authentication.proxy.user.identity.header'];
-      userid = req.headers[userIdProperty];
-
-      if (allowedOrigin.indexOf(req.headers.origin) === -1) {
+      if (!isAllowedOrigin(req)) {
         log.info('Unknown Origin: ' + req.headers.origin);
         log.info('Denying socket connection and closing the channel');
         socket.end();
         socket.destroy();
         return;
       }
+      socketIdentities.capture(req);
+    });
+    // Non-websocket sockjs transports (xhr-streaming, xhr-polling, eventsource, ...) never
+    // fire 'upgrade' -- they're plain HTTP requests, so this is the only place their
+    // PROXY-mode identity can be captured. Browsers don't reliably send Origin on
+    // same-origin requests, so a missing Origin here is treated as same-origin (matches
+    // how these fallback transports actually behave) rather than rejected outright; skip
+    // capture (rather than tearing down the socket, which sockjs's own request handling
+    // already owns) only for a mismatched, known-cross-origin request.
+    server.addListener('request', function(req) {
+      if (!isAllowedOrigin(req, true)) {
+        return;
+      }
+      socketIdentities.capture(req);
     });
     function gracefulShutdown() {
       log.info('Caught SIGTERM. Closing http & ws server');

server/socket-identity.mjs

@@ -0,0 +1,100 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * 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.
+ */
+
+/**
+ * In PROXY authentication mode, the auth token / user id for a sockjs
+ * session need to be captured from the raw HTTP request that establishes
+ * that session (see server.js for the full journey). A sockjs session isn't
+ * always established over a websocket upgrade though: whenever a websocket
+ * upgrade isn't available (some corporate proxies, older browsers), sockjs
+ * transparently falls back to plain HTTP transports (xhr-streaming,
+ * xhr-polling, eventsource, ...), which never fire Node's 'upgrade' event.
+ *
+ * This store keys captured identities by sockjs session id (parsed out of
+ * the request URL) instead of a single shared variable, since a shared
+ * variable gets clobbered by whichever request happens to arrive last --
+ * silently binding one user's session to a different, unrelated user's
+ * identity.
+ */
+
+/**
+ * @param {string} prefix sockjs mount prefix, e.g. '/_sock'
+ * @param {string} url request URL, shaped /<prefix>/<server>/<session>/<transport>[/...]
+ * @returns {string|null} the sockjs session id, or null if the URL isn't a sockjs request
+ */
+export function getSockjsSessionId(prefix, url) {
+  if (!url || url.indexOf(prefix + '/') !== 0) {
+    return null;
+  }
+  const parts = url.slice(prefix.length + 1).split('/');
+  return parts.length >= 2 && parts[1] ? parts[1] : null;
+}
+
+/**
+ * @param {object} opts
+ * @param {string} opts.prefix sockjs mount prefix, e.g. '/_sock'
+ * @param {object} opts.cdapConfig
+ * @param {(req: object) => string} opts.getAuthHeaderFromRawCookies
+ * @param {number} [opts.ttlMs] how long a captured identity survives if the session never connects
+ */
+export function createSocketIdentityStore({ prefix, cdapConfig, getAuthHeaderFromRawCookies, ttlMs = 60000 }) {
+  const pending = new Map();
+
+  /**
+   * Call for every request that might be starting/continuing a sockjs session.
+   * Long-lived polling transports call this repeatedly (once per poll) for the same
+   * session id, so each call clears the previous entry's timer before scheduling a new
+   * one -- otherwise an earlier, still-pending timer can delete a later, unconsumed entry.
+   */
+  function capture(req) {
+    const sessionId = getSockjsSessionId(prefix, req.url);
+    if (!sessionId) {
+      return;
+    }
+    const existing = pending.get(sessionId);
+    if (existing) {
+      clearTimeout(existing.timer);
+    }
+    req.headers.authorization = getAuthHeaderFromRawCookies(req);
+    const userIdProperty = cdapConfig['security.authentication.proxy.user.identity.header'];
+    const timer = setTimeout(() => pending.delete(sessionId), ttlMs);
+    if (typeof timer.unref === 'function') {
+      timer.unref();
+    }
+    pending.set(sessionId, {
+      authToken: req.headers.authorization,
+      userid: req.headers[userIdProperty],
+      timer,
+    });
+  }
+
+  /** Call once a sockjs 'connection' fires, with that connection's own url. */
+  function consume(url) {
+    const sessionId = getSockjsSessionId(prefix, url);
+    if (!sessionId) {
+      return {};
+    }
+    const identity = pending.get(sessionId);
+    if (!identity) {
+      return {};
+    }
+    clearTimeout(identity.timer);
+    pending.delete(sessionId);
+    return { authToken: identity.authToken, userid: identity.userid };
+  }
+
+  return { capture, consume };
+}

server/socket-identity.test.js

@@ -0,0 +1,124 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * 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.
+ */
+
+import { getSockjsSessionId, createSocketIdentityStore } from 'server/socket-identity';
+
+describe('getSockjsSessionId', () => {
+  test('extracts the session id from a sockjs transport URL', () => {
+    expect(getSockjsSessionId('/_sock', '/_sock/000/abc123/websocket')).toBe('abc123');
+    expect(getSockjsSessionId('/_sock', '/_sock/000/abc123/xhr_streaming')).toBe('abc123');
+  });
+
+  test('returns null for URLs outside the sockjs prefix', () => {
+    expect(getSockjsSessionId('/_sock', '/api/v3/namespaces')).toBeNull();
+    expect(getSockjsSessionId('/_sock', null)).toBeNull();
+  });
+});
+
+describe('createSocketIdentityStore', () => {
+  const cdapConfig = { 'security.authentication.proxy.user.identity.header': 'x-identity-id' };
+  const getAuthHeaderFromRawCookies = (req) => req.headers.authorization || '';
+
+  function makeStore() {
+    return createSocketIdentityStore({ prefix: '/_sock', cdapConfig, getAuthHeaderFromRawCookies });
+  }
+
+  test('binds a session to the identity captured from its own request', () => {
+    const store = makeStore();
+    store.capture({
+      url: '/_sock/000/mysession/websocket',
+      headers: { authorization: 'Bearer real-token', 'x-identity-id': 'real-user' },
+    });
+
+    const identity = store.consume('/_sock/000/mysession/websocket');
+    expect(identity.authToken).toBe('Bearer real-token');
+    expect(identity.userid).toBe('real-user');
+  });
+
+  // Regression test for the underlying bug: server.js used to keep a single
+  // shared `authToken`/`userid` variable, set inside the raw http 'upgrade'
+  // listener and read back inside sockjs's 'connection' handler. Because
+  // that state was shared across every in-flight session rather than scoped
+  // per session, a second, unrelated user's request could silently
+  // overwrite it before the first user's connection ever read it back --
+  // and any session established over a non-websocket transport (which never
+  // fires 'upgrade' at all) always read back whatever was last left behind
+  // by someone else's websocket connection.
+  test('does not leak one session identity into a concurrently-captured session', () => {
+    const store = makeStore();
+
+    // victim (e.g. an admin) starts a session
+    store.capture({
+      url: '/_sock/000/victim-session/websocket',
+      headers: { authorization: 'Bearer VICTIM-ADMIN-TOKEN', 'x-identity-id': 'victim-admin' },
+    });
+    // before the victim's 'connection' event is processed, an unrelated,
+    // concurrently-connecting user's request arrives (this is exactly what
+    // used to clobber the old shared variable)
+    store.capture({
+      url: '/_sock/000/attacker-session/xhr_streaming',
+      headers: { authorization: 'Bearer ATTACKER-TOKEN', 'x-identity-id': 'attacker-lowpriv' },
+    });
+
+    const victimIdentity = store.consume('/_sock/000/victim-session/websocket');
+    const attackerIdentity = store.consume('/_sock/000/attacker-session/xhr_streaming');
+
+    expect(victimIdentity.userid).toBe('victim-admin');
+    expect(victimIdentity.authToken).toBe('Bearer VICTIM-ADMIN-TOKEN');
+    expect(attackerIdentity.userid).toBe('attacker-lowpriv');
+    expect(attackerIdentity.authToken).toBe('Bearer ATTACKER-TOKEN');
+  });
+
+  test('a session with no matching capture (e.g. non-sockjs URL) consumes to an empty identity', () => {
+    const store = makeStore();
+    expect(store.consume('/_sock/000/never-captured/websocket')).toEqual({});
+  });
+
+  test('consuming a session removes it, so it cannot be read twice', () => {
+    const store = makeStore();
+    store.capture({
+      url: '/_sock/000/onceonly/websocket',
+      headers: { authorization: 'Bearer t', 'x-identity-id': 'u' },
+    });
+    store.consume('/_sock/000/onceonly/websocket');
+    expect(store.consume('/_sock/000/onceonly/websocket')).toEqual({});
+  });
+
+  // Regression test: a long-lived polling transport calls capture() repeatedly for the
+  // same session id. Each call used to schedule its own expiry timer without clearing the
+  // previous one, so an earlier timer could delete a later, still-unconsumed entry.
+  test('an earlier capture for a session does not expire a later capture of the same session', () => {
+    jest.useFakeTimers();
+    const store = createSocketIdentityStore({
+      prefix: '/_sock', cdapConfig, getAuthHeaderFromRawCookies, ttlMs: 1000,
+    });
+
+    store.capture({
+      url: '/_sock/000/polling/xhr_streaming',
+      headers: { authorization: 'Bearer first', 'x-identity-id': 'u' },
+    });
+    jest.advanceTimersByTime(900);
+    store.capture({
+      url: '/_sock/000/polling/xhr_streaming',
+      headers: { authorization: 'Bearer second', 'x-identity-id': 'u' },
+    });
+    // the first capture's timer would have fired here if it hadn't been cleared
+    jest.advanceTimersByTime(200);
+
+    expect(store.consume('/_sock/000/polling/xhr_streaming').authToken).toBe('Bearer second');
+    jest.useRealTimers();
+  });
+});