fix: support tls over cluster transport

src/ingress/http-ingress.ts

@@ -5,7 +5,7 @@ import EventBus from '../cluster/eventbus.js';
 import Listener from '../listener/listener.js';
 import IngressUtils from './utils.js';
 import { Logger } from '../logger.js';
-import TunnelService from '../tunnel/tunnel-service.js';
+import TunnelService, { CreateConnectionContext } from '../tunnel/tunnel-service.js';
 import AltNameService from '../tunnel/altname-service.js';
 import Node from '../cluster/cluster-node.js';
 import { ERROR_TUNNEL_NOT_FOUND,
@@ -186,13 +186,11 @@ export default class HttpIngress implements IngressBase {
 
         const remoteAddr = this._clientIp(req);
         const createConnection = (opts: object, callback: (err: Error | undefined, sock: Duplex) => void) => {
-            const ctx = {
+            const ctx: CreateConnectionContext = {
                 remoteAddr,
                 ingress: {
-                    tls: false,
                     port: this.httpListener.getPort(),
                 },
-                opts,
             };
             return this.tunnelService.createConnection(tunnelId, ctx, callback);
         };
@@ -425,10 +423,9 @@ export default class HttpIngress implements IngressBase {
             return true;
         }
 
-        const ctx = {
+        const ctx: CreateConnectionContext = {
             remoteAddr: this._clientIp(req),
             ingress: {
-                tls: false,
                 port: this.httpListener.getPort(),
             }
         };

src/ingress/sni-ingress.ts

@@ -302,7 +302,11 @@ export default class SNIIngress implements IngressBase {
         const ctx: CreateConnectionContext = {
             remoteAddr: socket.remoteAddress || '',
             ingress: {
-                tls: true,
+                tls: {
+                    enabled: true,
+                    servername,
+                    cert: this.cert,
+                },
                 port: this.port,
             },
         };

src/transport/cluster/cluster-transport.ts

@@ -1,5 +1,6 @@
 import { Duplex } from "stream";
-import { Socket, TcpSocketConnectOpts } from "net";
+import tls from "tls";
+import net from "net";
 import Transport, { TransportConnectionOptions, TransportOptions } from "../transport.js";
 import ClusterService from "../../cluster/index.js";
 
@@ -17,26 +18,47 @@ export default class ClusterTransport extends Transport {
     }
 
     public createConnection(opts: TransportConnectionOptions, callback: (err: Error | undefined, sock: Duplex) => void): Duplex {
+
         const clusterNode = this.clusterService.getNode(this.nodeId);
-        const sock = new Socket();
         if (!clusterNode) {
+            const sock = new net.Socket();
             sock.destroy(new Error('node_does_not_exist'));
             return sock;
         }
 
-        const socketOpts: TcpSocketConnectOpts = {
-            host: clusterNode.ip,
-            port: opts.port || 0,
-        };
+        let sock: tls.TLSSocket | net.Socket;
 
         const errorHandler = (err: Error) => {
             callback(err, sock);
         };
-        sock.once('error', errorHandler); 
-        sock.connect(socketOpts, () => {
-            sock.off('error', errorHandler);
-            callback(undefined, sock);
-        });
+
+        if (opts.tls?.enabled == true) {
+            const tlsConnectOpts: tls.ConnectionOptions = {
+                servername: opts.tls.servername,
+                host: clusterNode.ip,
+                port: opts.port || 0,
+                ca: [
+                    <any>opts.tls?.cert?.toString(),
+                    ...tls.rootCertificates,
+                ],
+            };
+            sock = tls.connect(tlsConnectOpts, () => {
+                sock.off('error', errorHandler);
+                callback(undefined, sock);
+            });
+            sock.once('error', errorHandler);
+        } else {
+            const socketConnectOpts: net.TcpSocketConnectOpts = {
+                host: clusterNode.ip,
+                port: opts.port || 0,
+            };
+            sock = net.connect(socketConnectOpts, () => {
+                sock.off('error', errorHandler);
+                callback(undefined, sock);
+            });
+            sock.once('error', errorHandler);
+        }
+
         return sock;
     }
 

src/transport/transport.ts

@@ -1,11 +1,18 @@
-import { randomUUID } from 'crypto';
+import crypto from 'crypto';
 import { EventEmitter } from 'events';
 import { Duplex } from 'stream';
 
+type TransportConnectTlsOptions = {
+    enabled: boolean,
+    servername?: string,
+    cert?: Buffer,
+};
+
 export type TransportConnectionOptions = {
     remoteAddr: string,
     tunnelId?: string,
     port?: number,
+    tls?: TransportConnectTlsOptions,
 };
 
 export interface TransportOptions {
@@ -20,7 +27,7 @@ export default abstract class Transport extends EventEmitter {
     constructor(opts: TransportOptions) {
         super();
         this.max_connections = opts.max_connections || 1;
-        this.id = randomUUID();
+        this.id = crypto.randomUUID();
     }
 
     public abstract createConnection(opts: TransportConnectionOptions, callback: (err: Error | undefined, sock: Duplex) => void): Duplex;

src/tunnel/tunnel-service.ts

@@ -21,11 +21,17 @@ export type ConnectOptions = {
     peer: string,
 }
 
+type CreateConnectionIngressTlsContext = {
+    enabled: boolean,
+    servername?: string,
+    cert?: Buffer,
+};
+
 export type CreateConnectionContext = {
     remoteAddr: string,
     ingress: {
         port: number,
-        tls?: boolean,
+        tls?: CreateConnectionIngressTlsContext,
     }
 };
 
@@ -796,6 +802,11 @@ export default class TunnelService {
             tunnelId,
             remoteAddr: ctx.remoteAddr,
             port: ctx.ingress.port,
+            tls: {
+                enabled: ctx.ingress.tls?.enabled == true,
+                servername: ctx.ingress.tls?.servername,
+                cert: ctx.ingress.tls?.cert,
+            }
         }, callback);
         return sock;
     }

test/e2e/test_cluster.js

@@ -2,8 +2,10 @@ import child_process from 'child_process';
 import crypto from 'crypto';
 import assert from 'assert/strict';
 import http from 'http';
+import https from 'https';
 import { setTimeout } from 'timers/promises';
-import { createAccount, createEchoServer, exposrCliImageTag, getAuthToken, getTunnel, putTunnel } from './e2e-utils.js';
+import { createAccount, exposrCliImageTag, getAuthToken, getTunnel, putTunnel } from './e2e-utils.js';
+import { createEchoHttpServer } from '../unit/test-utils.js';
 
 const startExposrd = (name = "", network, args = [], dockerargs = []) => {
     const obj = child_process.spawn("docker", [
@@ -80,39 +82,46 @@ describe('Cluster E2E', () => {
     });
 
     const clusterModes = [
-        {mode: "UDP/multicast", args: ["--cluster", "udp"]},
-        {mode: "Redis pub/sub", args: ["--cluster", "redis", "--cluster-redis-url", redisUrl, ]}
+        {mode: "UDP/multicast", ingress: 'http', args: ["--cluster", "udp"]},
+        {mode: "UDP/multicast", ingress: 'sni', args: ["--cluster", "udp"]},
+        {mode: "Redis pub/sub", ingress: 'http', args: ["--cluster", "redis", "--cluster-redis-url", redisUrl ]},
+        {mode: "Redis pub/sub", ingress: 'sni', args: ["--cluster", "redis", "--cluster-redis-url", redisUrl ]},
     ];
 
     // Test will
     // Spawn two nodes, with the given cluster method
     // Connect a tunnel client to the second node
-    // Perform a http ingress request to the first node 
+    // Perform a http ingress request to the first node
     // Assert that a http response is received
-    clusterModes.forEach(({mode, args}) => {
-        it(`Cluster mode ${mode} w/ redis storage`, async () => {
+    clusterModes.forEach(({mode, ingress, args}) => {
+        it(`Cluster mode ${mode} w/ redis storage, ingress ${ingress}`, async () => {
             const node1 = startExposrd("node-1", network, [
                 "--log-level", "debug",
-                "--storage-url", redisUrl, 
+                "--storage-url", redisUrl,
                 "--allow-registration",
-                "--ingress", "http",
+                "--ingress", "http,sni",
                 "--ingress-http-url", "http://localhost:8080",
+                "--ingress-sni-cert", "test/unit/fixtures/cn-public-cert.pem",
+                "--ingress-sni-key", "test/unit/fixtures/cn-private-key.pem",
             ].concat(args), [
-                "-p", "8080:8080"
+                "-p", "8080:8080",
+                "-p", "4430:4430"
             ]);
 
             const node2 = startExposrd("node-2", network, [
                 "--log-level", "debug",
-                "--storage-url", redisUrl, 
+                "--storage-url", redisUrl,
                 "--allow-registration",
-                "--ingress", "http",
+                "--ingress", "http,sni",
                 "--ingress-http-url", "http://localhost:8080",
+                "--ingress-sni-cert", "test/unit/fixtures/cn-public-cert.pem",
+                "--ingress-sni-key", "test/unit/fixtures/cn-private-key.pem",
             ].concat(args));
 
-            const echoServerTerminate = await createEchoServer();
+            const echoServer = await createEchoHttpServer();
 
             const apiEndpoint = "http://localhost:8080";
-            const echoServerUrl = "http://host.docker.internal:10000";
+            const echoServerUrl = "http://host.docker.internal:20000";
 
             let retries = 60;
             do {
@@ -131,50 +140,74 @@ describe('Cluster E2E', () => {
             const exposrCliTerminator = startExposr(
                 'http://node-2:8080', network, [
                 "-a", `${account.account_id}`,
-                "tunnel", "connect", `${tunnelId}`, `${echoServerUrl}`
+                "tunnel", "connect", `${tunnelId}`, `${echoServerUrl}`,
+                "ingress-http", "enable",
+                "ingress-sni", "enable",
             ]);
 
             authToken = await getAuthToken(account.account_id, apiEndpoint);
             let res, data;
             do {
                 await setTimeout(1000);
                 res = await getTunnel(authToken, tunnelId, apiEndpoint);
-                data = await res.json(); 
+                data = await res.json();
             } while (data?.connection?.connected == false);
 
             assert(data?.connection?.connected == true, "tunnel not connected");
 
             const ingressUrl = new URL(data.ingress.http.url);
+            const sniIngressUrl = new URL(data.ingress.sni.url);
 
             let status;
-            ([status, data] = await new Promise((resolve) => {
-                const req = http.request({
-                    hostname: 'localhost',
-                    port: 8080,
-                    method: 'POST',
-                    path: '/',
-                    headers: {
-                        "Host": ingressUrl.hostname
-                    }
-                }, (res) => {
+            ([status, data] = await new Promise((resolve, reject) => {
+                const onRes = (res) => {
                     let data = '';
-
                     res.on('data', (chunk) => {
                         data += chunk;
                     });
 
                     res.on('close', () => { resolve([res.statusCode, data])});
-                });
+                };
+
+                let req;
+                if (ingress == 'sni') {
+                    req = https.request({
+                        hostname: 'localhost',
+                        port: 4430,
+                        method: 'POST',
+                        path: '/',
+                        headers: {
+                            "Host": sniIngressUrl.hostname
+                        },
+                        servername: sniIngressUrl.hostname,
+                        rejectUnauthorized: false,
+                    }, onRes);
+                } else {
+                    req = http.request({
+                        hostname: 'localhost',
+                        port: 8080,
+                        method: 'POST',
+                        path: '/',
+                        headers: {
+                            "Host": ingressUrl.hostname
+                        },
+                        rejectUnauthorized: false,
+                    }, onRes);
+                }
+                req.on('error', (err) => {
+                    console.log(err);
+                    reject(err)
+                })
                 req.end('echo');
             }));
 
             exposrCliTerminator();
-            await echoServerTerminate();
             node1.terminate();
             node2.terminate();
+            await echoServer.destroy();
 
             assert(status == 200, `expected status code 200, got ${status}`);
             assert(data == "echo", `did not get response from echo server through WS tunnel, got ${data}`);
         }).timeout(120000);
-    }); 
+    });
 });