feat(microservice): add TLS over TCP support

packages/microservices/client/client-proxy-factory.ts

@@ -3,6 +3,7 @@ import {
   ClientOptions,
   CustomClientOptions,
   TcpClientOptions,
+  TcpTlsClientOptions,
 } from '../interfaces/client-metadata.interface';
 import { Closeable } from '../interfaces/closeable.interface';
 import {
@@ -56,13 +57,23 @@ export class ClientProxyFactory {
       case Transport.KAFKA:
         return new ClientKafka(options as KafkaOptions['options']);
       default:
-        return new ClientTCP(options as TcpClientOptions['options']);
+        const uncheckedOptions = options as
+          | TcpClientOptions['options']
+          | TcpTlsClientOptions['options']
+          | undefined;
+        if (uncheckedOptions && uncheckedOptions.useTls === true) {
+          return new ClientTCP(options as TcpTlsClientOptions['options']);
+        } else {
+          return new ClientTCP(
+            options as TcpClientOptions['options'] | undefined,
+          );
+        }
     }
   }
 
   private static isCustomClientOptions(
-    options: ClientOptions | CustomClientOptions,
+    options: ClientOptions | CustomClientOptions | undefined | null,
   ): options is CustomClientOptions {
-    return !!(options as CustomClientOptions).customClass;
+    return options && !!(options as CustomClientOptions).customClass;
   }
 }

packages/microservices/client/client-proxy.ts

@@ -25,6 +25,7 @@ import {
   RedisOptions,
   RmqOptions,
   TcpClientOptions,
+  TcpTlsClientOptions,
   WritePacket,
 } from '../interfaces';
 import { ProducerDeserializer } from '../interfaces/deserializer.interface';
@@ -128,7 +129,7 @@ export abstract class ClientProxy {
 
   protected getOptionsProp<
     T extends ClientOptions['options'],
-    K extends keyof T
+    K extends keyof T,
   >(obj: T, prop: K, defaultValue: T[K] = undefined) {
     return (obj && obj[prop]) || defaultValue;
   }
@@ -140,26 +141,32 @@ export abstract class ClientProxy {
   protected initializeSerializer(options: ClientOptions['options']) {
     this.serializer =
       (options &&
-        (options as
-          | RedisOptions['options']
-          | NatsOptions['options']
-          | MqttOptions['options']
-          | TcpClientOptions['options']
-          | RmqOptions['options']
-          | KafkaOptions['options']).serializer) ||
+        (
+          options as
+            | RedisOptions['options']
+            | NatsOptions['options']
+            | MqttOptions['options']
+            | TcpClientOptions['options']
+            | TcpTlsClientOptions['options']
+            | RmqOptions['options']
+            | KafkaOptions['options']
+        ).serializer) ||
       new IdentitySerializer();
   }
 
   protected initializeDeserializer(options: ClientOptions['options']) {
     this.deserializer =
       (options &&
-        (options as
-          | RedisOptions['options']
-          | NatsOptions['options']
-          | MqttOptions['options']
-          | TcpClientOptions['options']
-          | RmqOptions['options']
-          | KafkaOptions['options']).deserializer) ||
+        (
+          options as
+            | RedisOptions['options']
+            | NatsOptions['options']
+            | MqttOptions['options']
+            | TcpClientOptions['options']
+            | TcpTlsClientOptions['options']
+            | RmqOptions['options']
+            | KafkaOptions['options']
+        ).deserializer) ||
       new IncomingResponseDeserializer();
   }
 }

packages/microservices/client/client-tcp.ts

@@ -1,32 +1,55 @@
 import { Logger } from '@nestjs/common';
 import * as net from 'net';
-import { EmptyError, lastValueFrom } from 'rxjs';
+import { EmptyError, lastValueFrom, Observable } from 'rxjs';
 import { share, tap } from 'rxjs/operators';
+import * as tls from 'tls';
 import {
   CLOSE_EVENT,
   ECONNREFUSED,
   ERROR_EVENT,
   MESSAGE_EVENT,
   TCP_DEFAULT_HOST,
   TCP_DEFAULT_PORT,
+  TCP_DEFAULT_USE_TLS,
+  TLS_CONNECT_EVENT,
 } from '../constants';
 import { JsonSocket } from '../helpers/json-socket';
 import { PacketId, ReadPacket, WritePacket } from '../interfaces';
-import { TcpClientOptions } from '../interfaces/client-metadata.interface';
+import {
+  TcpClientOptions,
+  TcpTlsClientOptions,
+} from '../interfaces/client-metadata.interface';
 import { ClientProxy } from './client-proxy';
 
 export class ClientTCP extends ClientProxy {
   protected connection: Promise<any>;
   private readonly logger = new Logger(ClientTCP.name);
   private readonly port: number;
   private readonly host: string;
+  private readonly useTls: boolean;
   private isConnected = false;
   private socket: JsonSocket;
 
-  constructor(options: TcpClientOptions['options']) {
+  /**
+   * The underling netSocket used by the TLS Socket
+   */
+  private netSocket: net.Socket | null = null;
+
+  constructor();
+  constructor(options: TcpClientOptions['options']);
+  constructor(options: TcpTlsClientOptions['options']);
+  constructor(
+    private readonly options?:
+      | TcpClientOptions['options']
+      | TcpTlsClientOptions['options'],
+  ) {
     super();
+    if (options === undefined) {
+      this.options = {};
+    }
     this.port = this.getOptionsProp(options, 'port') || TCP_DEFAULT_PORT;
     this.host = this.getOptionsProp(options, 'host') || TCP_DEFAULT_HOST;
+    this.useTls = this.getOptionsProp(options, 'useTls') || TCP_DEFAULT_USE_TLS;
 
     this.initializeSerializer(options);
     this.initializeDeserializer(options);
@@ -39,17 +62,35 @@ export class ClientTCP extends ClientProxy {
     this.socket = this.createSocket();
     this.bindEvents(this.socket);
 
-    const source$ = this.connect$(this.socket.netSocket).pipe(
-      tap(() => {
-        this.isConnected = true;
-        this.socket.on(MESSAGE_EVENT, (buffer: WritePacket & PacketId) =>
-          this.handleResponse(buffer),
-        );
-      }),
-      share(),
-    );
-
-    this.socket.connect(this.port, this.host);
+    let source$: Observable<any>;
+
+    if (this.useTls) {
+      this.netSocket.connect(this.port, this.host);
+      source$ = this.connect$(
+        this.socket.netSocket,
+        ERROR_EVENT,
+        TLS_CONNECT_EVENT,
+      ).pipe(
+        tap(() => {
+          this.isConnected = true;
+          this.socket.on(MESSAGE_EVENT, (buffer: WritePacket & PacketId) =>
+            this.handleResponse(buffer),
+          );
+        }),
+        share(),
+      );
+    } else {
+      source$ = this.connect$(this.socket.netSocket).pipe(
+        tap(() => {
+          this.isConnected = true;
+          this.socket.on(MESSAGE_EVENT, (buffer: WritePacket & PacketId) =>
+            this.handleResponse(buffer),
+          );
+        }),
+        share(),
+      );
+      this.socket.connect(this.port, this.host);
+    }
     this.connection = lastValueFrom(source$).catch(err => {
       if (err instanceof EmptyError) {
         return;
@@ -81,7 +122,20 @@ export class ClientTCP extends ClientProxy {
   }
 
   public createSocket(): JsonSocket {
-    return new JsonSocket(new net.Socket());
+    let socket: net.Socket | tls.TLSSocket = new net.Socket();
+
+    /**
+     * TLS enabled, "upgrade" the TCP Socket to TLS
+     */
+    if (this.useTls === true) {
+      /**
+       * Options are TcpTlsClientOptions
+       */
+      const options = this.options as TcpTlsClientOptions['options'];
+      this.netSocket = socket;
+      socket = tls.connect({ ...options, socket });
+    }
+    return new JsonSocket(socket);
   }
 
   public close() {

packages/microservices/constants.ts

@@ -2,6 +2,7 @@ import { ROUTE_ARGS_METADATA } from '@nestjs/common/constants';
 
 export const TCP_DEFAULT_PORT = 3000;
 export const TCP_DEFAULT_HOST = 'localhost';
+export const TCP_DEFAULT_USE_TLS = false;
 export const REDIS_DEFAULT_URL = 'redis://localhost:6379';
 export const NATS_DEFAULT_URL = 'nats://localhost:4222';
 export const MQTT_DEFAULT_URL = 'mqtt://localhost:1883';
@@ -17,6 +18,7 @@ export const ERROR_EVENT = 'error';
 export const CLOSE_EVENT = 'close';
 export const SUBSCRIBE = 'subscribe';
 export const CANCEL_EVENT = 'cancelled';
+export const TLS_CONNECT_EVENT = 'secureConnect';
 
 export const PATTERN_METADATA = 'microservices:pattern';
 export const TRANSPORT_METADATA = 'microservices:transport';

packages/microservices/interfaces/client-metadata.interface.ts

@@ -11,6 +11,7 @@ import {
   RmqOptions,
 } from './microservice-configuration.interface';
 import { Serializer } from './serializer.interface';
+import { TLSSocketOptions } from 'tls';
 
 export type ClientOptions =
   | RedisOptions
@@ -19,19 +20,27 @@ export type ClientOptions =
   | GrpcOptions
   | KafkaOptions
   | TcpClientOptions
+  | TcpTlsClientOptions
   | RmqOptions;
 
 export interface CustomClientOptions {
   customClass: Type<ClientProxy>;
   options?: Record<string, any>;
 }
 
+export interface TcpClientBaseOptions {
+  host?: string;
+  port?: number;
+  serializer?: Serializer;
+  deserializer?: Deserializer;
+}
+
 export interface TcpClientOptions {
   transport: Transport.TCP;
-  options?: {
-    host?: string;
-    port?: number;
-    serializer?: Serializer;
-    deserializer?: Deserializer;
-  };
+  options?: (TcpClientBaseOptions & { useTls?: false | undefined }) | undefined;
+}
+
+export interface TcpTlsClientOptions {
+  transport: Transport.TCP;
+  options: TcpClientBaseOptions & { useTls: true } & TLSSocketOptions;
 }

packages/microservices/interfaces/microservice-configuration.interface.ts

@@ -1,3 +1,4 @@
+import { TlsOptions } from 'tls';
 import { Transport } from '../enums/transport.enum';
 import { ChannelOptions } from '../external/grpc-options.interface';
 import {
@@ -18,6 +19,7 @@ import { Serializer } from './serializer.interface';
 export type MicroserviceOptions =
   | GrpcOptions
   | TcpOptions
+  | TcpTlsOptions
   | RedisOptions
   | NatsOptions
   | MqttOptions
@@ -67,16 +69,23 @@ export interface GrpcOptions {
   };
 }
 
+interface TcpBaseOptions {
+  host?: string;
+  port?: number;
+  retryAttempts?: number;
+  retryDelay?: number;
+  serializer?: Serializer;
+  deserializer?: Deserializer;
+}
+
 export interface TcpOptions {
   transport?: Transport.TCP;
-  options?: {
-    host?: string;
-    port?: number;
-    retryAttempts?: number;
-    retryDelay?: number;
-    serializer?: Serializer;
-    deserializer?: Deserializer;
-  };
+  options?: TcpBaseOptions & { useTls?: false | undefined };
+}
+
+export interface TcpTlsOptions {
+  transport?: Transport.TCP;
+  options: TcpBaseOptions & { useTls: true } & TlsOptions;
 }
 
 export interface RedisOptions {

packages/microservices/server/server-tcp.ts

@@ -1,5 +1,6 @@
 import { isString, isUndefined } from '@nestjs/common/utils/shared.utils';
 import * as net from 'net';
+import * as tls from 'tls';
 import { Server as NetSocket, Socket } from 'net';
 import { Observable } from 'rxjs';
 import {
@@ -11,6 +12,7 @@ import {
   NO_MESSAGE_HANDLER,
   TCP_DEFAULT_HOST,
   TCP_DEFAULT_PORT,
+  TCP_DEFAULT_USE_TLS,
 } from '../constants';
 import { TcpContext } from '../ctx-host/tcp.context';
 import { Transport } from '../enums';
@@ -22,22 +24,31 @@ import {
   ReadPacket,
   WritePacket,
 } from '../interfaces';
-import { TcpOptions } from '../interfaces/microservice-configuration.interface';
+import {
+  TcpOptions,
+  TcpTlsOptions,
+} from '../interfaces/microservice-configuration.interface';
 import { Server } from './server';
 
 export class ServerTCP extends Server implements CustomTransportStrategy {
   public readonly transportId = Transport.TCP;
 
   private readonly port: number;
   private readonly host: string;
+  private readonly useTls: boolean;
   private server: NetSocket;
   private isExplicitlyTerminated = false;
   private retryAttemptsCount = 0;
 
-  constructor(private readonly options: TcpOptions['options']) {
+  constructor(options: TcpOptions['options']);
+  constructor(options: TcpTlsOptions['options']);
+  constructor(
+    private readonly options: TcpOptions['options'] | TcpTlsOptions['options'],
+  ) {
     super();
     this.port = this.getOptionsProp(options, 'port') || TCP_DEFAULT_PORT;
     this.host = this.getOptionsProp(options, 'host') || TCP_DEFAULT_HOST;
+    this.useTls = this.getOptionsProp(options, 'useTls') || TCP_DEFAULT_USE_TLS;
 
     this.init();
     this.initializeSerializer(options);
@@ -119,7 +130,14 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
   }
 
   private init() {
-    this.server = net.createServer(this.bindHandler.bind(this));
+    if (this.useTls) {
+      // TLS enabled, use tls server
+      const options = this.options as TcpTlsOptions['options'];
+      this.server = tls.createServer(options, this.bindHandler.bind(this));
+    } else {
+      // TLS disabled, use net server
+      this.server = net.createServer(this.bindHandler.bind(this));
+    }
     this.server.on(ERROR_EVENT, this.handleError.bind(this));
     this.server.on(CLOSE_EVENT, this.handleClose.bind(this));
   }