feat(repeater): refrain from utilizing non-standard ports

package.json

@@ -40,6 +40,8 @@
     "request": "~2.88.2",
     "request-promise": "~4.2.6",
     "semver": "^7.3.5",
+    "socket.io-client": "^4.7.1",
+    "socket.io-msgpack-parser": "^3.0.2",
     "socks": "~2.6.1",
     "socks-proxy-agent": "~5.0.0",
     "tslib": "^2.3.1",

src/Commands/RunRepeater.ts

@@ -1,8 +1,7 @@
-import { RabbitMQBusOptions } from '../Bus';
 import { Cert, RequestExecutorOptions } from '../RequestExecutor';
 import { Helpers, logger } from '../Utils';
 import { container } from '../Config';
-import { RepeaterLauncher } from '../Repeater';
+import { DefaultRepeaterServerOptions, RepeaterLauncher } from '../Repeater';
 import { Arguments, Argv, CommandModule } from 'yargs';
 import { normalize } from 'path';
 
@@ -178,19 +177,17 @@ export class RunRepeater implements CommandModule {
               ]
             }
           })
-          .register<RabbitMQBusOptions>(RabbitMQBusOptions, {
-            useValue: {
-              exchange: 'EventBus',
-              clientQueue: `agent:${args.id as string}`,
-              connectTimeout: 10000,
-              url: args.bus as string,
-              proxyUrl: (args.proxyExternal ?? args.proxy) as string,
-              credentials: {
-                username: 'bot',
-                password: args.token as string
+          .register<DefaultRepeaterServerOptions>(
+            DefaultRepeaterServerOptions,
+            {
+              useValue: {
+                uri: args.repeaterServer as string,
+                token: args.token as string,
+                connectTimeout: 10000,
+                proxyUrl: (args.proxyExternal ?? args.proxy) as string
               }
             }
-          })
+          )
       );
   }
 

src/Config/CliBuilder.ts

@@ -64,9 +64,11 @@ export class CliBuilder {
         describe: 'SOCKS4 or SOCKS5 URL to proxy internal traffic'
       })
       .middleware((args: Arguments) => {
-        ({ bus: args.bus, api: args.api } = Helpers.getClusterUrls(
-          args as ClusterArgs
-        ));
+        ({
+          bus: args.bus,
+          api: args.api,
+          repeaterServer: args.repeaterServer
+        } = Helpers.getClusterUrls(args as ClusterArgs));
       })
       // TODO: (victor.polyakov@brightsec.com) Write correct type checking
       .middleware(

src/Config/container.ts

@@ -53,7 +53,12 @@ import { ConfigReader } from './ConfigReader';
 import { DefaultConfigReader } from './DefaultConfigReader';
 import { CliInfo } from './CliInfo';
 import { CliBuilder } from './CliBuilder';
-import { DefaultRepeaterLauncher, RepeaterLauncher } from '../Repeater';
+import {
+  ServerRepeaterLauncher,
+  RepeaterLauncher,
+  RepeaterServer,
+  DefaultRepeaterServer
+} from '../Repeater';
 import { container, Lifecycle } from 'tsyringe';
 
 container
@@ -103,6 +108,13 @@ container
     },
     { lifecycle: Lifecycle.Singleton }
   )
+  .register(
+    RepeaterServer,
+    {
+      useClass: DefaultRepeaterServer
+    },
+    { lifecycle: Lifecycle.Singleton }
+  )
   .register(
     Tokens,
     {
@@ -202,7 +214,7 @@ container
   )
   .register<RepeaterLauncher>(
     RepeaterLauncher,
-    { useClass: DefaultRepeaterLauncher },
+    { useClass: ServerRepeaterLauncher },
     { lifecycle: Lifecycle.Singleton }
   )
   .register<CliBuilder>(CliBuilder, {

src/Repeater/DefaultRepeaterServer.ts

@@ -0,0 +1,207 @@
+import { logger } from '../Utils';
+import {
+  RepeaterServer,
+  RepeaterServerDeployedEvent,
+  RepeaterServerReconnectionFailedEvent,
+  RepeaterServerRequestEvent,
+  RepeaterServerRequestResponse,
+  RepeaterServerErrorEvent,
+  RepeaterServerReconnectionAttemptedEvent
+} from './RepeaterServer';
+import { inject, injectable } from 'tsyringe';
+import io, { Socket } from 'socket.io-client';
+import parser from 'socket.io-msgpack-parser';
+import { SocksProxyAgent } from 'socks-proxy-agent';
+import { once } from 'events';
+import { parse } from 'url';
+import Timer = NodeJS.Timer;
+
+export interface DefaultRepeaterServerOptions {
+  readonly uri: string;
+  readonly token: string;
+  readonly connectTimeout?: number;
+  readonly proxyUrl?: string;
+}
+
+export const DefaultRepeaterServerOptions: unique symbol = Symbol(
+  'DefaultRepeaterServerOptions'
+);
+
+@injectable()
+export class DefaultRepeaterServer implements RepeaterServer {
+  private latestReconnectionError?: Error;
+  private readonly MAX_RECONNECTION_ATTEMPTS = 20;
+  private _socket?: Socket;
+  private timer?: Timer;
+
+  constructor(
+    @inject(DefaultRepeaterServerOptions)
+    private readonly options: DefaultRepeaterServerOptions
+  ) {}
+
+  public disconnect() {
+    this.clearPingTimer();
+
+    this._socket?.disconnect();
+    this._socket?.removeAllListeners();
+    this._socket = undefined;
+  }
+
+  public async deploy(
+    repeaterId?: string
+  ): Promise<RepeaterServerDeployedEvent> {
+    this.socket.emit('deploy', {
+      repeaterId
+    });
+
+    const [result]: RepeaterServerDeployedEvent[] = await once(
+      this.socket,
+      'deployed'
+    );
+
+    return result;
+  }
+
+  public connect(hostname: string) {
+    this._socket = io(this.options.uri, {
+      parser,
+      path: '/api/ws/v1',
+      transports: ['websocket'],
+      timeout: this.options?.connectTimeout,
+      // @ts-expect-error Type is wrong.
+      // Agent is passed directly to "ws" package, which accepts http.Agent
+      agent: this.options.proxyUrl
+        ? new SocksProxyAgent({
+            ...parse(this.options.proxyUrl)
+          })
+        : false,
+      reconnectionAttempts: this.MAX_RECONNECTION_ATTEMPTS,
+      auth: {
+        token: this.options.token,
+        domain: hostname
+      }
+    });
+
+    this.socket.on('connect_error', (error: Error) => {
+      logger.debug(`Unable to connect to the %s host`, this.options.uri, error);
+    });
+
+    this.createPingTimer();
+
+    logger.debug('Event bus connected to %s', this.options.uri);
+  }
+
+  public requestReceived(
+    handler: (
+      event: RepeaterServerRequestEvent
+    ) => RepeaterServerRequestResponse | Promise<RepeaterServerRequestResponse>
+  ): void {
+    this.socket.on('request', (payload, callback) => {
+      this.processEventHandler('request', payload, handler, callback);
+    });
+  }
+
+  public reconnectionFailed(
+    handler: (
+      event: RepeaterServerReconnectionFailedEvent
+    ) => void | Promise<void>
+  ): void {
+    this.socket.io.on('reconnect', () => {
+      this.latestReconnectionError = undefined;
+    });
+
+    this.socket.io.on('reconnect_error', (error) => {
+      this.latestReconnectionError = error;
+    });
+
+    this.socket.io.on('reconnect_failed', () => {
+      this.processEventHandler(
+        'reconnection_failed',
+        {
+          error: this.latestReconnectionError
+        },
+        handler
+      );
+    });
+  }
+
+  public errorOccurred(
+    handler: (event: RepeaterServerErrorEvent) => void | Promise<void>
+  ): void {
+    this.socket.on('error', (payload, callback) => {
+      this.processEventHandler('error', payload, handler, callback);
+    });
+  }
+
+  public reconnectionAttempted(
+    handler: (
+      event: RepeaterServerReconnectionAttemptedEvent
+    ) => void | Promise<void>
+  ): void {
+    this.socket.io.on('reconnect_attempt', (attempt) => {
+      this.processEventHandler(
+        'reconnect_attempt',
+        { attempt, maxAttempts: this.MAX_RECONNECTION_ATTEMPTS },
+        handler
+      );
+    });
+  }
+
+  public reconnectionSucceeded(handler: () => void | Promise<void>): void {
+    this.socket.io.on('reconnect', () => {
+      this.processEventHandler('reconnect', undefined, handler);
+    });
+  }
+
+  private get socket() {
+    if (!this._socket) {
+      throw new Error(
+        'Please make sure that repeater established a connection with host.'
+      );
+    }
+
+    return this._socket;
+  }
+
+  private processEventHandler<P>(
+    event: string,
+    payload: P,
+    handler: (payload: P) => unknown,
+    callback?: unknown
+  ) {
+    (async function () {
+      try {
+        const response = await handler(payload);
+
+        if (typeof callback !== 'function') {
+          return;
+        }
+
+        callback(response);
+      } catch (error) {
+        logger.debug(
+          'Error processing event "%s" with the following payload: %s. Details: %s',
+          event,
+          payload,
+          error
+        );
+        logger.error('Error: %s', error.message);
+      }
+    })();
+  }
+
+  private createPingTimer() {
+    this.clearPingTimer();
+
+    this.timer = setInterval(
+      () => this.socket.volatile.emit('ping'),
+      10000
+    ).unref();
+  }
+
+  private clearPingTimer() {
+    if (this.timer) {
+      clearInterval(this.timer);
+    }
+  }
+}

src/Repeater/RepeaterLauncher.ts

@@ -3,7 +3,10 @@ export interface RepeaterLauncher {
 
   loadScripts(scripts: Record<string, string>): Promise<void>;
 
-  compileScripts(scripts: string | Record<string, string>): void;
+  /**
+   * @deprecated currently not supported by some implementations
+   */
+  compileScripts?(scripts: string | Record<string, string>): void;
 
   run(repeaterId: string, asDaemon?: boolean): Promise<void>;
 

src/Repeater/RepeaterServer.ts

@@ -0,0 +1,76 @@
+import { Protocol } from '../RequestExecutor';
+
+export interface RepeaterServerDeployedEvent {
+  repeaterId: string;
+}
+
+export interface RepeaterServerRequestEvent {
+  protocol: Protocol;
+  url: string;
+  method?: string;
+  headers?: Record<string, string | string[]>;
+  correlationIdRegex?: string;
+  body?: string;
+}
+
+export type RepeaterServerRequestResponse =
+  | {
+      protocol: Protocol;
+      statusCode?: number;
+      message?: string;
+      errorCode?: string;
+      headers?: Record<string, string | string[] | undefined>;
+      body?: string;
+    }
+  | {
+      protocol: Protocol;
+      message?: string;
+      errorCode?: string;
+    };
+
+export interface RepeaterServerReconnectionFailedEvent {
+  error: Error;
+}
+
+export interface RepeaterServerReconnectionAttemptedEvent {
+  attempt: number;
+  maxAttempts: number;
+}
+
+export interface RepeaterServerErrorEvent {
+  message: string;
+}
+
+export interface RepeaterServer {
+  disconnect(): void;
+
+  connect(hostname: string): void;
+
+  deploy(repeaterId?: string): Promise<RepeaterServerDeployedEvent>;
+
+  requestReceived(
+    handler: (
+      event: RepeaterServerRequestEvent
+    ) => RepeaterServerRequestResponse | Promise<RepeaterServerRequestResponse>
+  ): void;
+
+  reconnectionFailed(
+    handler: (
+      event: RepeaterServerReconnectionFailedEvent
+    ) => void | Promise<void>
+  ): void;
+
+  reconnectionAttempted(
+    handler: (
+      event: RepeaterServerReconnectionAttemptedEvent
+    ) => void | Promise<void>
+  ): void;
+
+  reconnectionSucceeded(handler: () => void | Promise<void>): void;
+
+  errorOccurred(
+    handler: (event: RepeaterServerErrorEvent) => void | Promise<void>
+  ): void;
+}
+
+export const RepeaterServer: unique symbol = Symbol('RepeaterServer');