diff --git a/server.js b/server.js index 659ee3354e5..a9ab4dc927c 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,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'); diff --git a/server/socket-identity.mjs b/server/socket-identity.mjs new file mode 100644 index 00000000000..b603af4d79d --- /dev/null +++ b/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 ////[/...] + * @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 }; +} diff --git a/server/socket-identity.test.js b/server/socket-identity.test.js new file mode 100644 index 00000000000..17017f1a995 --- /dev/null +++ b/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(); + }); +});