feat: add server side connection auth management

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/server/auth/authHelpers.ts

@@ -0,0 +1,50 @@
+import { getConfig } from '../../common/config/config';
+import { PostFilterPreparedRequest } from '../../common/relay/prepareRequest';
+import { maskToken } from '../../common/utils/token';
+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),
+  };
+  logger.debug(
+    { maskToken: maskToken(brokerConnectionIdentifier) },
+    `Validate Broker Client Credentials request`,
+  );
+  const response = await makeSingleRawRequestToDownstream(req);
+  logger.debug(
+    { validationResponseCode: response.statusCode },
+    'Validate Broker Client Credentials response',
+  );
+  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/auth/connectionWatchdog.ts

@@ -0,0 +1,30 @@
+import { getConfig } from '../../common/config/config';
+import { getSocketConnections } from '../socket';
+import { log as logger } from '../../logs/logger';
+
+export const disconnectConnectionsWithStaleCreds = async () => {
+  const connections = getSocketConnections();
+  const connectionsIterator = connections.entries();
+  for (const [identifier, connection] of connectionsIterator) {
+    connection.forEach((client) => {
+      if (!isDateWithinAnHourAndFiveSec(client.credsValidationTime!)) {
+        logger.debug(
+          {
+            connection: `${identifier}`,
+            credsLastValidated: client.credsValidationTime,
+          },
+          'Cutting off connection.',
+        );
+        client.socket!.end();
+      }
+    });
+  }
+};
+
+const isDateWithinAnHourAndFiveSec = (date: string): boolean => {
+  const dateInMs = new Date(date); // Convert ISO string to Date
+  const now = Date.now(); // Get current time in milliseconds
+  const staleConnectionsCleanupInterval =
+    getConfig().STALE_CONNECTIONS_CLEANUP_FREQUENCY ?? 65 * 60 * 1000; // 1h05 hour in milliseconds
+  return now - dateInMs.getTime() < staleConnectionsCleanupInterval;
+};

lib/server/index.ts

@@ -13,6 +13,8 @@ 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';
+import { disconnectConnectionsWithStaleCreds } from './auth/connectionWatchdog';
 
 export const main = async (serverOpts: ServerOpts) => {
   logger.info({ version }, 'Broker starting in server mode');
@@ -64,7 +66,24 @@ export const main = async (serverOpts: ServerOpts) => {
     getForwardHttpRequestHandler(),
   );
 
-  app.post('/response-data/:brokerToken/:streamingId', handlePostResponse);
+  if (loadedServerOpts.config.BROKER_SERVER_MANDATORY_AUTH_ENABLED) {
+    app.post(
+      '/hidden/brokers/connections/:identifier/auth/refresh',
+      authRefreshHandler,
+    );
+    app.post(
+      '/hidden/broker/response-data/:brokerToken/:streamingId',
+      handlePostResponse,
+    );
+
+    setInterval(
+      disconnectConnectionsWithStaleCreds,
+      loadedServerOpts.config.STALE_CONNECTIONS_CLEANUP_FREQUENCY ??
+        10 * 60 * 1000,
+    );
+  } 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,88 @@
+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';
+import { maskToken } from '../../common/utils/token';
+interface BrokerConnectionAuthRequest {
+  data: {
+    attributes: {
+      broker_client_id: string;
+    };
+    id: string;
+    type: 'broker_connection';
+  };
+}
+export const authRefreshHandler = async (req: Request, res: Response) => {
+  const credentialsFromHeader =
+    req.headers['Authorization'] ?? req.headers['authorization'];
+  const role = req.query['connection_role'];
+  const credentials = `${credentialsFromHeader}`;
+  const brokerAppClientId =
+    req.headers[`${process.env.SNYK_INTERNAL_AUTH_CLIENT_ID_HEADER}`];
+  const identifier = req.params.identifier;
+  logger.debug(
+    { maskedToken: maskToken(identifier), brokerAppClientId, role },
+    `Auth Refresh`,
+  );
+  const body = JSON.parse(req.body.toString()) as BrokerConnectionAuthRequest;
+  const brokerClientId = body.data.attributes.broker_client_id;
+  if (!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 && x.role === role,
+      )
+    : null;
+  logger.debug({ identifier, brokerClientId, role }, 'Validating credentials');
+  if (
+    credentials === undefined ||
+    brokerAppClientId === undefined ||
+    !connection ||
+    !currentClient
+  ) {
+    logger.debug(
+      { identifier, brokerClientId, role, credentials },
+      'Invalid credentials',
+    );
+    return res.status(401).send('Invalid credentials.');
+  } else {
+    const credsCheckResponse = await validateBrokerClientCredentials(
+      credentials,
+      brokerClientId as string,
+      identifier,
+    );
+    logger.debug(
+      { credsCheckResponse: credsCheckResponse },
+      'Client Creds validation response.',
+    );
+    if (credsCheckResponse) {
+      // Refresh client validation time
+      const nowDate = new Date().toISOString();
+      currentClient.credsValidationTime = nowDate;
+      const currentClientIndex = connection.findIndex(
+        (x) => x.brokerClientId === brokerClientId && x.role === role,
+      );
+      if (currentClientIndex > -1) {
+        connection[currentClientIndex] = currentClient;
+        return res.status(201).send('OK');
+      } else {
+        return res.status(500).send('Unable to find client connection.');
+      }
+    } else {
+      logger.debug(
+        { identifier, brokerClientId, role, credentials },
+        'Invalid credentials - Creds check response returned false',
+      );
+      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')) {

lib/server/routesHandlers/postResponseHandler.ts

@@ -4,6 +4,8 @@ import { log as logger } from '../../logs/logger';
 import { getDesensitizedToken } from '../utils/token';
 import { incrementHttpRequestsTotal } from '../../common/utils/metrics';
 import { StreamResponseHandler } from '../../hybrid-sdk/http/server-post-stream-handler';
+import { getConfig } from '../../common/config/config';
+import { decode } from 'jsonwebtoken';
 
 export const handlePostResponse = (req: Request, res: Response) => {
   incrementHttpRequestsTotal(false, 'data-response');
@@ -32,6 +34,35 @@ export const handlePostResponse = (req: Request, res: Response) => {
       .json({ message: 'unable to find request matching streaming id' });
     return;
   }
+  if (getConfig().BROKER_SERVER_MANDATORY_AUTH_ENABLED) {
+    const credentials = req.headers.authorization;
+    if (!credentials) {
+      logger.error(
+        logContext,
+        'Invalid Broker Client credentials on response data',
+      );
+      res.status(401).json({ message: 'Invalid Broker Client credentials' });
+      return;
+    }
+    const decodedJwt = credentials
+      ? decode(credentials!.replace(/bearer /i, ''), {
+          complete: true,
+        })
+      : null;
+
+    const brokerAppClientId = decodedJwt ? decodedJwt?.payload['azp'] : '';
+    if (
+      !brokerAppClientId ||
+      brokerAppClientId != streamHandler.streamResponse.brokerAppClientId
+    ) {
+      logger.error(
+        logContext,
+        'Invalid Broker Client credentials for stream on response data',
+      );
+      res.status(401).json({ message: 'Invalid Broker Client credentials' });
+      return;
+    }
+  }
   let statusAndHeaders = '';
   let statusAndHeadersSize = -1;