From 96ac65bd39b5fc344b899d07ef5bbdc4b4a588fb Mon Sep 17 00:00:00 2001 From: Herdiyan Adam Putra Date: Wed, 5 Aug 2026 20:46:30 +0700 Subject: [PATCH 1/2] server: fix cross-user identity binding race in PROXY auth mode websocket setup server.js captured the sockjs-session auth token / user id (used to make backend API calls under PROXY authentication mode) in 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, concurrently-connecting user's request could clobber it before an earlier session's 'connection' handler read it back -- binding that earlier session to the wrong user's identity. Sessions established over a non-websocket sockjs transport (xhr-streaming, xhr-polling, eventsource -- sockjs's normal fallback whenever a websocket upgrade isn't available) never fire 'upgrade' at all, so they always read back whatever the last unrelated websocket connection had left behind. Verified with the real sockjs/ws packages: 50 concurrent websocket connections against the old logic produced 50/50 cross-bound identities; an xhr-streaming session deterministically inherited a previous, unrelated websocket session's auth token and user id. Extracted the capture/lookup into server/socket-identity.mjs, keyed by sockjs session id (parsed from the request URL) instead of one shared variable, and wired it into both the 'upgrade' and 'request' listeners so every transport is covered. Also moved the origin check ahead of identity capture in the 'upgrade' listener, since the old ordering let a rejected-origin request still populate the shared state before being denied. --- server.js | 51 ++++++++++++++---- server/socket-identity.mjs | 86 +++++++++++++++++++++++++++++ server/socket-identity.test.js | 99 ++++++++++++++++++++++++++++++++++ 3 files changed, 225 insertions(+), 11 deletions(-) create mode 100644 server/socket-identity.mjs create mode 100644 server/socket-identity.test.js diff --git a/server.js b/server.js index 659ee3354e5..7e8564172e0 100644 --- a/server.js +++ b/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,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; + } + sockServer.on('connection', function(c) { if (!c) { log.error('Connection requested, but no connection available'); @@ -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)); @@ -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); }); function gracefulShutdown() { log.info('Caught SIGTERM. Closing http & ws server'); diff --git a/server/socket-identity.mjs b/server/socket-identity.mjs new file mode 100644 index 00000000000..51b71ca2cad --- /dev/null +++ b/server/socket-identity.mjs @@ -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 ////[/...] + * @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 }; +} diff --git a/server/socket-identity.test.js b/server/socket-identity.test.js new file mode 100644 index 00000000000..f314f97fbd7 --- /dev/null +++ b/server/socket-identity.test.js @@ -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({}); + }); +}); From 622b70bc067a0962dc3af2c5fa6e3476d8f4ad00 Mon Sep 17 00:00:00 2001 From: Herdiyan Adam Putra Date: Wed, 5 Aug 2026 20:54:59 +0700 Subject: [PATCH 2/2] Address review feedback: allow missing Origin on fallback transports, fix timer leak/race isAllowedOrigin now takes an allowMissing flag. Browsers don't reliably send Origin on same-origin requests, and the new 'request' listener had no equivalent check before this PR at all, so treating a missing Origin as same-origin there (matches the fallback transports' actual behavior) avoids breaking PROXY-mode xhr-streaming/xhr-polling sessions outright. The websocket upgrade path keeps the strict, no-missing-origin check it already had. createSocketIdentityStore now clears the previous entry's timer before scheduling a new one. Polling transports call capture() once per poll for the same session id; without this, an earlier call's timer could fire and delete a later, still-unconsumed entry. --- server.js | 18 ++++++++++++------ server/socket-identity.mjs | 28 +++++++++++++++++++++------- server/socket-identity.test.js | 25 +++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 13 deletions(-) diff --git a/server.js b/server.js index 7e8564172e0..a9ab4dc927c 100644 --- a/server.js +++ b/server.js @@ -240,8 +240,12 @@ getCDAPConfig() getAuthHeaderFromRawCookies, }); - function isAllowedOrigin(req) { - return allowedOrigin.indexOf(req.headers.origin) !== -1; + function isAllowedOrigin(req, allowMissing = false) { + const origin = req.headers.origin; + if (!origin) { + return allowMissing; + } + return allowedOrigin.indexOf(origin) !== -1; } sockServer.on('connection', function(c) { @@ -278,11 +282,13 @@ getCDAPConfig() }); // 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. + // 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)) { + if (!isAllowedOrigin(req, true)) { return; } socketIdentities.capture(req); diff --git a/server/socket-identity.mjs b/server/socket-identity.mjs index 51b71ca2cad..b603af4d79d 100644 --- a/server/socket-identity.mjs +++ b/server/socket-identity.mjs @@ -53,22 +53,32 @@ export function getSockjsSessionId(prefix, url) { export function createSocketIdentityStore({ prefix, cdapConfig, getAuthHeaderFromRawCookies, ttlMs = 60000 }) { const pending = new Map(); - /** Call for every request that might be starting/continuing a sockjs session. */ + /** + * 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']; - 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(); } + pending.set(sessionId, { + authToken: req.headers.authorization, + userid: req.headers[userIdProperty], + timer, + }); } /** Call once a sockjs 'connection' fires, with that connection's own url. */ @@ -77,9 +87,13 @@ export function createSocketIdentityStore({ prefix, cdapConfig, getAuthHeaderFro if (!sessionId) { return {}; } - const identity = pending.get(sessionId) || {}; + const identity = pending.get(sessionId); + if (!identity) { + return {}; + } + clearTimeout(identity.timer); pending.delete(sessionId); - return identity; + return { authToken: identity.authToken, userid: identity.userid }; } return { capture, consume }; diff --git a/server/socket-identity.test.js b/server/socket-identity.test.js index f314f97fbd7..17017f1a995 100644 --- a/server/socket-identity.test.js +++ b/server/socket-identity.test.js @@ -96,4 +96,29 @@ describe('createSocketIdentityStore', () => { 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(); + }); });