Feat/add server side new auth flow [HYB-725]

accept-server.local.json

@@ -0,0 +1,18 @@
+{
+    "//": "private refers to what's internal to snyk, i.e. the snyk.io server",
+    "private": [
+        {
+            "//": "send any type of request to our connected clients",
+            "method": "any",
+            "path": "/*"
+        }
+    ],
+    "public": [
+        {
+            "//": "send any type of request to our connected clients",
+            "method": "any",
+            "path": "/*"
+        }
+    ]
+}
+

lib/client/auth/brokerServerConnection.ts

@@ -0,0 +1,50 @@
+import { getConfig } from '../../common/config/config';
+import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import version from '../../common/utils/version';
+import {
+  HttpResponse,
+  makeRequestToDownstream,
+} from '../../hybrid-sdk/http/request';
+import { Role } from '../types/client';
+
+export interface BrokerServerConnectionParams {
+  connectionIdentifier: string;
+  brokerClientId: string;
+  authorization: string;
+  role: Role;
+  serverId: number;
+}
+export const renewBrokerServerConnection = async (
+  brokerServerConnectionParams: BrokerServerConnectionParams,
+): Promise<HttpResponse> => {
+  const clientConfig = getConfig();
+  const apiHostname = clientConfig.API_BASE_URL;
+  const body = {
+    data: {
+      type: 'broker_connection',
+      attributes: {
+        broker_client_id: brokerServerConnectionParams.brokerClientId,
+      },
+    },
+  };
+  const url = new URL(
+    `${apiHostname}/hidden/brokers/connections/${brokerServerConnectionParams.connectionIdentifier}/auth/refresh`,
+  );
+  url.searchParams.append('role', brokerServerConnectionParams.role);
+  if (brokerServerConnectionParams.serverId) {
+    url.searchParams.append(
+      'serverId',
+      `${brokerServerConnectionParams.serverId}`,
+    );
+  }
+  const req: PostFilterPreparedRequest = {
+    url: url.toString(),
+    headers: {
+      authorization: brokerServerConnectionParams.authorization,
+      'user-agent': `Snyk Broker Client ${version}`,
+    },
+    method: 'POST',
+    body: JSON.stringify(body),
+  };
+  return await makeRequestToDownstream(req);
+};

lib/client/auth/oauth.ts

@@ -38,7 +38,6 @@ export async function fetchJwt(
     const jwt = accessToken.access_token;
     const type = accessToken.token_type;
     const expiresIn = accessToken.expires_in;
-
     return { expiresIn: expiresIn, authHeader: `${type} ${jwt}` };
   } catch (err) {
     logger.error({ err }, 'Unable to retrieve JWT');

lib/client/index.ts

@@ -89,6 +89,7 @@ export const main = async (clientOpts: ClientOpts) => {
         clientOpts.config.brokerClientId,
       );
     }
+
     await retrieveAndLoadFilters(clientOpts);
     if (process.env.NODE_ENV != 'test') {
       setInterval(async () => {

lib/client/socket.ts

@@ -22,6 +22,7 @@ import { fetchJwt } from './auth/oauth';
 import { getServerId } from './dispatcher';
 import { determineFilterType } from './utils/filterSelection';
 import { notificationHandler } from './socketHandlers/notificationHandler';
+import { renewBrokerServerConnection } from './auth/brokerServerConnection';
 
 export const createWebSocketConnectionPairs = async (
   websocketConnections: WebSocketConnection[],
@@ -66,7 +67,6 @@ export const createWebSocketConnectionPairs = async (
   } else {
     logger.info(
       {
-        connection: socketIdentifyingMetadata.friendlyName,
         serverId: serverId,
       },
       'received server id',
@@ -97,6 +97,11 @@ export const createWebSocket = (
   const localClientOps = Object.assign({}, clientOpts);
   identifyingMetadata.identifier =
     identifyingMetadata.identifier ?? localClientOps.config.brokerToken;
+  if (!identifyingMetadata.identifier) {
+    throw new Error(
+      `Invalid Broker Identifier/Token in websocket tunnel creation step.`,
+    );
+  }
   const Socket = Primus.createSocket({
     transformer: 'engine.io',
     parser: 'EJSON',
@@ -175,8 +180,18 @@ export const createWebSocket = (
 
       websocket.transport.extraHeaders['Authorization'] =
         clientOpts.accessToken!.authHeader;
-      // websocket.end();
-      // websocket.open();
+      if (clientOpts.config.WS_TUNNEL_BOUNCE_ON_AUTH_REFRESH) {
+        websocket.end();
+        websocket.open();
+      } else {
+        await renewBrokerServerConnection({
+          connectionIdentifier: identifyingMetadata.identifier!,
+          brokerClientId: identifyingMetadata.clientId,
+          authorization: clientOpts.accessToken!.authHeader,
+          role: identifyingMetadata.role,
+          serverId: serverId,
+        });
+      }
       timeoutHandlerId = setTimeout(
         timeoutHandler,
         (clientOpts.accessToken!.expiresIn - 60) * 1000,
@@ -235,6 +250,26 @@ export const createWebSocket = (
     openHandler(websocket, localClientOps, identifyingMetadata),
   );
 
+  websocket.on('service', (msg) => {
+    logger.info({ msg }, 'service message received');
+  });
+  // websocket.on('outgoing::open', function () {
+  //   type OnErrorHandler = (type: string, code: number) => void;
+
+  //   const originalErrorHandler: OnErrorHandler =
+  //     websocket.socket.transport.onError;
+
+  //   websocket.socket.transport.onError = (...args: [string, number]) => {
+  //     const [type, code] = args; // Destructure for clarity
+  //     if (code === 401) {
+  //       logger.error({ type, code }, `Connection denied: unauthorized.`);
+  //     } else {
+  //       logger.error({ type, code }, `Transport error during polling.`);
+  //     }
+  //     originalErrorHandler.apply(websocket.socket?.transport, args);
+  //   };
+  // });
+
   websocket.on('close', () =>
     closeHandler(localClientOps, identifyingMetadata),
   );

lib/hybrid-sdk/clientRequestHelpers.ts

@@ -82,6 +82,7 @@ export class HybridClientRequestHandler {
       response: this.res,
       streamBuffer,
       streamSize: 0,
+      brokerAppClientId: this.res.locals.brokerAppClientId,
     });
     streamBuffer.pipe(this.res);
     const simplifiedContextWithStreamingID = this.simplifiedContext;

lib/hybrid-sdk/http/downstream-post-stream-to-server.ts

@@ -64,8 +64,11 @@ class BrokerServerPostResponseHandler {
 
   async #initHttpClientRequest() {
     try {
+      const backendHostname = this.#config.universalBrokerEnabled
+        ? `${this.#config.API_BASE_URL}/hidden/broker`
+        : this.#config.brokerServerUrl;
       const url = new URL(
-        `${this.#config.brokerServerUrl}/response-data/${this.#brokerToken}/${
+        `${backendHostname}/response-data/${this.#brokerToken}/${
           this.#streamingId
         }`,
       );

lib/hybrid-sdk/http/server-post-stream-handler.ts

@@ -15,6 +15,7 @@ export interface StreamResponse {
   streamBuffer: stream.PassThrough;
   response: Response;
   streamSize?: number;
+  brokerAppClientId: string;
 }
 
 export class StreamResponseHandler {
@@ -35,12 +36,18 @@ export class StreamResponseHandler {
       streamingID,
       streamResponse.streamBuffer,
       streamResponse.response,
+      streamResponse.brokerAppClientId,
     );
   }
 
-  constructor(streamingID, streamBuffer, response) {
+  constructor(streamingID, streamBuffer, response, brokerAppClientId) {
     this.streamingID = streamingID;
-    this.streamResponse = { streamBuffer, response, streamSize: 0 };
+    this.streamResponse = {
+      streamBuffer,
+      response,
+      streamSize: 0,
+      brokerAppClientId,
+    };
   }
 
   writeStatusAndHeaders = (statusAndHeaders) => {

lib/server/auth/authHelpers.ts

@@ -0,0 +1,42 @@
+import { getConfig } from '../../common/config/config';
+import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import { makeSingleRawRequestToDownstream } from '../../hybrid-sdk/http/request';
+import { log as logger } from '../../logs/logger';
+
+export const validateBrokerClientCredentials = async (
+  authHeaderValue: string,
+  brokerClientId: string,
+  brokerConnectionIdentifier: string,
+) => {
+  const body = {
+    data: {
+      type: 'broker_connection',
+      attributes: {
+        broker_client_id: brokerClientId,
+      },
+    },
+  };
+
+  const req: PostFilterPreparedRequest = {
+    url: `${
+      getConfig().apiHostname
+    }/hidden/brokers/connections/${brokerConnectionIdentifier}/auth/validate?version=2024-02-08~experimental`,
+    headers: {
+      authorization: authHeaderValue,
+      'Content-type': 'application/vnd.api+json',
+    },
+    method: 'POST',
+    body: JSON.stringify(body),
+  };
+
+  const response = await makeSingleRawRequestToDownstream(req);
+  if (response.statusCode === 201) {
+    return true;
+  } else {
+    logger.debug(
+      { statusCode: response.statusCode, message: response.statusText },
+      `Broker ${brokerConnectionIdentifier} client ID ${brokerClientId} failed validation.`,
+    );
+    return false;
+  }
+};

lib/server/index.ts

@@ -13,6 +13,7 @@ import { getForwardHttpRequestHandler } from './socketHandlers/initHandlers';
 import { loadAllFilters } from '../common/filter/filtersAsync';
 import { FiltersType } from '../common/types/filter';
 import filterRulesLoader from '../common/filter/filter-rules-loading';
+import { authRefreshHandler } from './routesHandlers/authHandlers';
 
 export const main = async (serverOpts: ServerOpts) => {
   logger.info({ version }, 'Broker starting in server mode');
@@ -57,14 +58,28 @@ export const main = async (serverOpts: ServerOpts) => {
     app.use(applyPrometheusMiddleware());
   }
   app.get('/connection-status/:token', connectionStatusHandler);
+  app.post(
+    '/hidden/brokers/connections/:identifier/auth/refresh',
+    authRefreshHandler,
+  );
   app.all(
     '/broker/:token/*',
     overloadHttpRequestWithConnectionDetailsMiddleware,
     validateBrokerTypeMiddleware,
     getForwardHttpRequestHandler(),
   );
 
-  app.post('/response-data/:brokerToken/:streamingId', handlePostResponse);
+  if (
+    loadedServerOpts.config.BROKER_SERVER_MANDATORY_AUTH_ENABLED ||
+    loadedServerOpts.config.RESPONSE_DATA_HIDDEN_ENABLED
+  ) {
+    app.post(
+      '/hidden/broker/response-data/:brokerToken/:streamingId',
+      handlePostResponse,
+    );
+  } else {
+    app.post('/response-data/:brokerToken/:streamingId', handlePostResponse);
+  }
 
   app.get('/', (req, res) => res.status(200).json({ ok: true, version }));
 

lib/server/routesHandlers/authHandlers.ts

@@ -0,0 +1,60 @@
+import { Request, Response } from 'express';
+import { validateBrokerClientCredentials } from '../auth/authHelpers';
+import { log as logger } from '../../logs/logger';
+import { validate } from 'uuid';
+import { getSocketConnectionByIdentifier } from '../socket';
+interface BrokerConnectionAuthRequest {
+  data: {
+    attributes: {
+      broker_client_id: string;
+    };
+    id: string;
+    type: 'broker_connection';
+  };
+}
+export const authRefreshHandler = async (req: Request, res: Response) => {
+  const credentials = req.headers['authorization'];
+  const brokerAppClientId =
+    req.headers[`${process.env.SNYK_INTERNAL_AUTH_CLIENT_ID_HEADER}`];
+  const identifier = req.params.identifier;
+  const body = JSON.parse(req.body.toString()) as BrokerConnectionAuthRequest;
+  const brokerClientId = body.data.attributes.broker_client_id;
+  if (
+    !validate(identifier) ||
+    !validate(brokerClientId) ||
+    !validate(brokerAppClientId)
+  ) {
+    logger.warn(
+      { identifier, brokerClientId, brokerAppClientId },
+      'Invalid credentials',
+    );
+    return res.status(401).send('Invalid parameters or credentials.');
+  }
+
+  const connection = getSocketConnectionByIdentifier(identifier);
+  const currentClient = connection
+    ? connection.find((x) => x.metadata.clientId === brokerClientId)
+    : null;
+  logger.debug({ identifier, brokerClientId }, 'Validating credentials');
+  if (
+    credentials === undefined ||
+    brokerAppClientId === undefined ||
+    credentials?.split('.').length != 3 ||
+    !connection ||
+    !currentClient
+  ) {
+    return res.status(401).send('Invalid credentials.');
+  } else {
+    const credsCheckResponse = await validateBrokerClientCredentials(
+      credentials,
+      brokerClientId as string,
+      identifier,
+    );
+    if (credsCheckResponse) {
+      return res.status(200).send('OK');
+    } else {
+      currentClient.socket!.end();
+      return res.status(401).send('Invalid credentials.');
+    }
+  }
+};

lib/server/routesHandlers/connectionStatusHandler.ts

@@ -11,7 +11,7 @@ export const connectionStatusHandler = async (req: Request, res: Response) => {
   const desensitizedToken = getDesensitizedToken(token);
   const connections = getSocketConnections();
   if (connections.has(token)) {
-    const clientsMetadata = connections.get(req.params.token).map((conn) => ({
+    const clientsMetadata = connections.get(req.params.token)!.map((conn) => ({
       version: conn.metadata && conn.metadata.version,
       filters: conn.metadata && conn.metadata.filters,
     }));

lib/server/routesHandlers/httpRequestHandler.ts

@@ -66,12 +66,14 @@ export const overloadHttpRequestWithConnectionDetailsMiddleware = async (
 
   // Grab a first (newest) client from the pool
   // This is really silly...
-  res.locals.websocket = connections.get(token)[0].socket;
-  res.locals.socketVersion = connections.get(token)[0].socketVersion;
-  res.locals.capabilities = connections.get(token)[0].metadata.capabilities;
+  res.locals.websocket = connections.get(token)![0].socket;
+  res.locals.socketVersion = connections.get(token)![0].socketVersion;
+  res.locals.capabilities = connections.get(token)![0].metadata.capabilities;
+  res.locals.brokerAppClientId =
+    connections.get(token)![0].brokerAppClientId ?? '';
   req['locals'] = {};
   req['locals']['capabilities'] =
-    connections.get(token)[0].metadata.capabilities;
+    connections.get(token)![0].metadata.capabilities;
   // strip the leading url
   req.url = req.url.slice(`/broker/${token}`.length);
   if (req.url.includes('connection_role')) {