Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 40 additions & 11 deletions server.js
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -215,14 +216,34 @@ 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) {
return allowedOrigin.indexOf(req.headers.origin) !== -1;
}

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

In same-origin HTTP requests (such as fallback transports like xhr-streaming or eventsource), browsers typically do not send an Origin header. Under the current implementation, isAllowedOrigin will return false when req.headers.origin is undefined, causing the request listener to skip capturing the session identity. This completely breaks fallback transports in PROXY mode.

We should update isAllowedOrigin to accept an optional allowMissing parameter (defaulting to false to preserve strict origin validation for WebSocket upgrades) and return true if the Origin header is missing and allowMissing is enabled.

    function isAllowedOrigin(req, allowMissing = false) {
      const origin = req.headers.origin;
      if (!origin) {
        return allowMissing;
      }
      return allowedOrigin.indexOf(origin) !== -1;
    }

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 622b70b. isAllowedOrigin now takes an allowMissing flag, and the request listener passes true since these fallback transports don't reliably send Origin on same-origin requests. Upgrade path stays strict.


sockServer.on('connection', function(c) {
if (!c) {
log.error('Connection requested, but no connection available');
Expand All @@ -231,8 +252,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));
Expand All @@ -243,20 +265,27 @@ 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. Skip capture (rather than tearing down the
// socket, which sockjs's own request handling already owns) for origins we wouldn't
// have allowed to upgrade either, so those sessions fall through unauthenticated.
server.addListener('request', function(req) {
if (!isAllowedOrigin(req)) {
return;
}
socketIdentities.capture(req);
});
Comment on lines +290 to 295

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

Update the request listener to call isAllowedOrigin(req, true) so that same-origin fallback transport requests without an Origin header are allowed to have their identities captured.

Suggested change
server.addListener('request', function(req) {
if (!isAllowedOrigin(req)) {
return;
}
socketIdentities.capture(req);
});
server.addListener('request', function(req) {
if (!isAllowedOrigin(req, true)) {
return;
}
socketIdentities.capture(req);
});

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in 622b70b, request listener now calls isAllowedOrigin(req, true).

function gracefulShutdown() {
log.info('Caught SIGTERM. Closing http & ws server');
Expand Down
86 changes: 86 additions & 0 deletions server/socket-identity.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
/*
* 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. */
function capture(req) {
const sessionId = getSockjsSessionId(prefix, req.url);
if (!sessionId) {
return;
}
req.headers.authorization = getAuthHeaderFromRawCookies(req);
const userIdProperty = cdapConfig['security.authentication.proxy.user.identity.header'];
pending.set(sessionId, {
authToken: req.headers.authorization,
userid: req.headers[userIdProperty],
});
const timer = setTimeout(() => pending.delete(sessionId), ttlMs);
if (typeof timer.unref === 'function') {
timer.unref();
}
}

/** 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) || {};
pending.delete(sessionId);
return identity;
}

return { capture, consume };
}
Comment on lines +53 to +100

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

There are two issues with the current timer management in createSocketIdentityStore:

  1. Dangling/Leaked Timeouts: Every call to capture (which happens on every polling request for fallback transports) schedules a new setTimeout that is never cleared, even after the identity is successfully consumed or overwritten. This leads to unnecessary CPU overhead and memory churn.
  2. Premature Deletion Race: If multiple handshake or transport requests for the same session ID occur before the connection is established, multiple timeouts are scheduled. When the first timeout fires, it will delete the session identity from pending, even if a subsequent request refreshed it.

Additionally, we should add defensive checks for cdapConfig to prevent runtime errors if it is null or undefined.

We can resolve these issues by storing the timer reference in the pending map, clearing any existing timeout when overwriting or consuming the session, and adding defensive checks.

export function createSocketIdentityStore({ prefix, cdapConfig, getAuthHeaderFromRawCookies, ttlMs = 60000 }) {
  const pending = new Map();

  /** Call for every request that might be starting/continuing a sockjs session. */
  function capture(req) {
    const sessionId = getSockjsSessionId(prefix, req.url);
    if (!sessionId) {
      return;
    }

    // Clear any existing timeout for this session to avoid premature deletion
    const existing = pending.get(sessionId);
    if (existing && existing.timer) {
      clearTimeout(existing.timer);
    }

    req.headers.authorization = getAuthHeaderFromRawCookies(req);
    const userIdProperty = cdapConfig && 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: userIdProperty ? req.headers[userIdProperty] : undefined,
      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 {};
    }
    if (identity.timer) {
      clearTimeout(identity.timer);
    }
    pending.delete(sessionId);
    return {
      authToken: identity.authToken,
      userid: identity.userid,
    };
  }

  return { capture, consume };
}

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fair point, polling transports call capture() once per poll for the same session so the old timers were piling up and one could delete a fresher entry before it got consumed. Fixed in 622b70b: the timer is now stored on the entry and cleared on overwrite/consume, plus added a fake-timers test for it.

99 changes: 99 additions & 0 deletions server/socket-identity.test.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
/*
* 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({});
});
});