[OGUI-1775] Create broadcast method for central system

Tokenization/backend/wrapper/package.json

@@ -26,6 +26,8 @@
   "dependencies": {
     "@grpc/grpc-js": "^1.13.4",
     "@grpc/proto-loader": "^0.7.15",
+    "express": "^5.1.0",
+    "jose": "^6.1.0",
     "ts-node": "^10.9.2",
     "typescript": "^5.8.3"
   }

Tokenization/backend/wrapper/src/central/CentralSystemWrapper.ts

@@ -15,6 +15,8 @@
 import * as grpc from '@grpc/grpc-js';
 import * as protoLoader from '@grpc/proto-loader';
 import { LogManager } from '@aliceo2/web-ui';
+import * as fs from 'fs';
+import type { CentralSystemConfig } from '../models/config.model';
 import type { DuplexMessageModel } from '../models/message.model';
 
 /**
@@ -24,11 +26,15 @@ export class CentralSystemWrapper {
   // Config
   private _protoPath: string;
 
-  // Utilities
-  private _logger = LogManager.getLogger('CentralSystemWrapper');
-
   // Class properties
   private _server: grpc.Server;
+  private _port: number;
+
+  // Certificates paths
+  private _serverCerts: CentralSystemConfig['serverCerts'];
+
+  // Utilities
+  private _logger = LogManager.getLogger('CentralSystemWrapper');
 
   // Clients management
   private _clients = new Map<string, grpc.ServerDuplexStream<any, any>>();
@@ -38,8 +44,15 @@ export class CentralSystemWrapper {
    * Initializes the Wrapper for CentralSystem.
    * @param port The port number to bind the gRPC server to.
    */
-  constructor(protoPath: string, private port: number) {
-    this._protoPath = protoPath;
+  constructor(config: CentralSystemConfig) {
+    if (!config.protoPath || !config.serverCerts?.caCertPath || !config.serverCerts?.certPath || !config.serverCerts?.keyPath) {
+      throw new Error('Invalid CentralSystemConfig provided');
+    }
+
+    this._protoPath = config.protoPath;
+    this._serverCerts = config.serverCerts;
+    this._port = config.port ?? 50051;
+
     this._server = new grpc.Server();
     this.setupService();
   }
@@ -115,7 +128,7 @@ export class CentralSystemWrapper {
 
     // Handle stream error event
     call.on('error', (err) => {
-      this._logger.infoMessage(`Stream error from client ${clientIp}:`, err);
+      this._logger.errorMessage(`Stream error from client ${clientIp}:`, err);
       this.cleanupClient(peer);
     });
   }
@@ -156,6 +169,17 @@ export class CentralSystemWrapper {
     }
   }
 
+  public broadcastEvent(data: DuplexMessageModel): void {
+    this.clients.forEach((client, ip) => {
+      try {
+        client.write(data);
+        this.logger.infoMessage(`Broadcasted event to ${ip}:`, data);
+      } catch (err) {
+        this.logger.errorMessage(`Error broadcasting to ${ip}:`, err);
+      }
+    });
+  }
+
   /**
    * Gets all connected client IPs
    * @returns Array of connected client IPs
@@ -168,10 +192,27 @@ export class CentralSystemWrapper {
    * Starts the gRPC server and binds it to the specified in class port.
    */
   public listen() {
-    const addr = `localhost:${this.port}`;
-    this._server.bindAsync(addr, grpc.ServerCredentials.createInsecure(), (err, _port) => {
+    const addr = `localhost:${this._port}`;
+
+    // Create mTLS secure gRPC server
+    const caCert = fs.readFileSync(this._serverCerts.caCertPath);
+    const centralKey = fs.readFileSync(this._serverCerts.keyPath);
+    const centralCert = fs.readFileSync(this._serverCerts.certPath);
+
+    const sslCreds = grpc.ServerCredentials.createSsl(
+      caCert,
+      [
+        {
+          private_key: centralKey,
+          cert_chain: centralCert,
+        },
+      ],
+      true
+    );
+
+    this._server.bindAsync(addr, sslCreds, (err, _port) => {
       if (err) {
-        this._logger.infoMessage('Server bind error:', err);
+        this._logger.errorMessage('Server bind error:', err);
         return;
       }
       this._logger.infoMessage(`CentralSytem started listening on ${addr}`);

Tokenization/backend/wrapper/src/client/connection/Connection.ts

@@ -12,60 +12,132 @@
  * or submit itself to any jurisdiction.
  */
 
-import type { ConnectionDirection } from '../../models/message.model';
-import type { ConnectionHeaders, FetchOptions, FetchResponse } from '../../models/connection.model';
+import { ConnectionDirection } from '../../models/message.model';
+import type { ConnectionHeaders, FetchOptions, FetchResponse, TokenPayload } from '../../models/connection.model';
 import { ConnectionStatus } from '../../models/connection.model';
 import * as grpc from '@grpc/grpc-js';
+import { LogManager } from '@aliceo2/web-ui';
+
+type ConnectionCerts = {
+  caCert: Buffer;
+  clientCert: Buffer;
+  clientKey: Buffer;
+};
 
 /**
  * @description This class represents a connection to a target client and manages sending messages to it.
  */
 export class Connection {
-  private _token: string;
-  private _targetAddress: string;
+  private _jweToken: string;
   private _status: ConnectionStatus;
-  private _peerClient: any;
+  private _peerClient?: any; // A client grpc connection instance
+
+  // Security management variables
+  private _clientSerialNumber?: string; // The certificate SN used to uniquely identify the peer.
+  private _lastActiveTimestamp: number; // Timestamp of the last successful request (for garbage collection).
+  private _authFailures: number; // Counter for consecutive authentication failures (for anti-DDoS/throttling).
+  private _cachedTokenPayload?: TokenPayload; // Cache of the successfully verified token payload.
+
+  public targetAddress: string;
   public direction: ConnectionDirection;
 
+  // Utils
+  private _logger;
+
   /**
    * Creates a new Connection instance with the given token, target address, and connection direction.
    *
-   * @param token - The authentication token for the connection.
+   * @param jweToken - The encrypted JWE token for the connection.
    * @param targetAddress - The unique address of the target client.
    * @param direction - The direction of the connection (e.g., sending or receiving).
+   * @param clientSN - Optional serial number of the peer's certificate (used for lookups).
    */
-  constructor(token: string, targetAddress: string, direction: ConnectionDirection, peerCtor: any) {
-    this._token = token;
-    this._targetAddress = targetAddress;
-    this._peerClient = new peerCtor(targetAddress, grpc.credentials.createInsecure());
+  constructor(jweToken: string, targetAddress: string, direction: ConnectionDirection, clientSN?: string) {
+    this._jweToken = jweToken;
+    this.targetAddress = targetAddress;
     this.direction = direction;
 
+    // Initialize state fields
+    this._clientSerialNumber = clientSN;
+    this._lastActiveTimestamp = Date.now();
+    this._authFailures = 0;
     this._status = ConnectionStatus.CONNECTED;
+
+    this._logger = LogManager.getLogger(`Connection ${targetAddress}`);
+  }
+
+  /**
+   * Creates the mTLS gRPC client and attaches it to the connection.
+   * This method is REQUIRED ONLY for outbound (SENDING) connections.
+   * * @param peerCtor - The constructor for the gRPC client to be used for communication.
+   * @param connectionCerts - Required sending client certificates for mTLS.
+   */
+  public createSslTunnel(peerCtor: any, connectionCerts: ConnectionCerts): void {
+    if (this.direction !== ConnectionDirection.SENDING) {
+      this._logger.warnMessage('Attempted to create SSL tunnel on a RECEIVING connection. This is usually unnecessary.');
+    }
+
+    if (!connectionCerts.caCert || !connectionCerts.clientCert || !connectionCerts.clientKey) {
+      throw new Error('Connection certificates are required to create an mTLS tunnel.');
+    }
+
+    // Create grpc credentials
+    const sslCreds = grpc.credentials.createSsl(connectionCerts.caCert, connectionCerts.clientKey, connectionCerts.clientCert);
+
+    this._peerClient = new peerCtor(this.targetAddress, sslCreds);
+    this.status = ConnectionStatus.CONNECTED;
   }
 
   /**
    * Replace newly generated token
-   * @param token New token to be replaced
+   * @param jweToken New token to be replaced
    */
-  public set token(token: string) {
-    this._token = token;
+  public set token(jweToken: string) {
+    this._jweToken = jweToken;
   }
 
   /**
-   * Revokes the token of the connection, effectively invalidating it.
-   * The connection status is set to UNAUTHORIZED.
+   * Revoke current token and set status of unauthorized connection
    */
   public handleRevokeToken(): void {
-    this._token = '';
-    this._status = ConnectionStatus.UNAUTHORIZED;
+    this._jweToken = '';
+    this.status = ConnectionStatus.UNAUTHORIZED;
+  }
+
+  /**
+   * Handles a successful authentication event. Updates the active timestamp,
+   * resets the failure counter, and caches the new token payload.
+   * This is crucial for high-performance applications to avoid re-validating the same token.
+   * @param payload The decoded and verified token payload.
+   */
+  public handleSuccessfulAuth(payload: TokenPayload): void {
+    this._lastActiveTimestamp = Date.now();
+    this._authFailures = 0;
+    this._cachedTokenPayload = payload;
+    this.status = ConnectionStatus.CONNECTED;
+  }
+
+  /**
+   * Handles an authentication failure. Increments the failure counter.
+   * If the failure count exceeds a local threshold, the connection is locally marked as BLOCKED.
+   * @returns The new count of consecutive failures.
+   */
+  public handleFailedAuth(): number {
+    this._authFailures += 1;
+
+    // Local throttling mechanism
+    if (this._authFailures >= 5) {
+      this.status = ConnectionStatus.BLOCKED;
+    }
+    return this._authFailures;
   }
 
   /**
    * Returns token for this Connection object
    * @returns Connection token
    */
   public get token(): string {
-    return this._token;
+    return this._jweToken;
   }
 
   /**
@@ -77,19 +149,51 @@ export class Connection {
   }
 
   /**
-   * Sets the status of this connection.
-   * @param status The new status of the connection.
+   * Updates the status of the connection.
+   * @param status New status
    */
   public set status(status: ConnectionStatus) {
     this._status = status;
   }
 
   /**
-   * Returns target address for this Connection object
-   * @returns Target address
+   * Returns the client's Serial Number (SN).
+   * @returns The client's serial number or undefined.
+   */
+  public get serialNumber(): string | undefined {
+    return this._clientSerialNumber;
+  }
+
+  /**
+   * Sets the client's Serial Number. Primarily used for RECEIVING connections
+   * where the SN is extracted during the first mTLS handshake in the interceptor.
+   * @param serialNumber The serial number string.
+   */
+  public set serialNumber(serialNumber: string | undefined) {
+    this._clientSerialNumber = serialNumber;
+  }
+
+  /**
+   * Returns the timestamp of the last successful interaction.
+   * @returns UNIX timestamp in milliseconds.
+   */
+  public get lastActiveTimestamp(): number {
+    return this._lastActiveTimestamp;
+  }
+
+  /**
+   * Returns the cached token payload.
+   * @returns The cached payload or undefined.
    */
-  public get targetAddress(): string {
-    return this._targetAddress;
+  public get cachedTokenPayload(): TokenPayload | undefined {
+    return this._cachedTokenPayload;
+  }
+
+  /**
+   * Attaches gRPC client to that connection
+   */
+  public attachGrpcClient(client: any): void {
+    this._peerClient = client;
   }
 
   /**
@@ -106,6 +210,11 @@ export class Connection {
     const path = options.path ?? '/';
     const headers: ConnectionHeaders = { ...(options.headers ?? {}) };
 
+    // Set mandatory grpc metadata
+    const metadata = new grpc.Metadata();
+    metadata.set('jwetoken', this._jweToken);
+
+    // Build body buffer
     let bodyBuf: Buffer = Buffer.alloc(0);
     const b = options.body;
     if (b != null) {
@@ -119,7 +228,7 @@ export class Connection {
 
     // Return promise with response
     return new Promise<FetchResponse>((resolve, reject) => {
-      this._peerClient.Fetch(req, (err: any, resp: any) => {
+      this._peerClient.Fetch(req, metadata, (err: any, resp: any) => {
         if (err) return reject(err);
 
         const resBody = resp?.body ? Buffer.from(resp.body) : Buffer.alloc(0);