Added user authentication & user termination

src/pusher/apps/app.ts

@@ -5,6 +5,8 @@ export abstract class App implements FN.Pusher.PusherApps.App {
     key: string;
     secret: string;
 
+    enableUserAuthentication?: boolean;
+    userAuthenticationTimeout?: number;
     enableClientMessages: boolean;
     enableMetrics: boolean;
     enabled: boolean;
@@ -53,6 +55,16 @@ export abstract class App implements FN.Pusher.PusherApps.App {
             default: 'app-secret',
             parameters: ['secret'],
         },
+        enableUserAuthentication: {
+            default: false,
+            parameters: ['enableUserAuthentication'],
+            parsers: [Boolean],
+        },
+        userAuthenticationTimeout: {
+            default: 10_000,
+            parameters: ['userAuthenticationTimeout'],
+            parsers: [parseInt],
+        },
         enableClientMessages: {
             default: false,
             parameters: ['enableClientMessages'],
@@ -192,12 +204,12 @@ export abstract class App implements FN.Pusher.PusherApps.App {
     abstract createToken(params: string): Promise<string>;
     abstract sha256(): Promise<string>;
 
-    async calculateSigningToken(
+    async calculateRequestToken(
         params: { [key: string]: string },
         method: string,
         path: string,
         body?: string,
-    ): Promise<string> {;
+    ): Promise<string> {
         params['auth_key'] = this.key;
 
         delete params['auth_signature'];
@@ -213,6 +225,14 @@ export abstract class App implements FN.Pusher.PusherApps.App {
         return await this.createToken([method, path, App.toOrderedArray(params).join('&')].join("\n"));
     }
 
+    async calculateSigninToken(connId: string, userData: string): Promise<string> {
+        return await this.createToken(`${connId}::user::${userData}`);
+    }
+
+    async signinTokenIsValid(connId: string, userData: string, receivedToken: string): Promise<boolean> {
+        return `${this.key}:${await this.calculateSigninToken(connId, userData)}` === receivedToken;
+    }
+
     protected transformPotentialJsonToArray(potentialJson: any): any {
         if (potentialJson instanceof Array) {
             return potentialJson;

src/pusher/channels/private-channel-manager.ts

@@ -29,13 +29,11 @@ export class PrivateChannelManager extends PublicChannelManager implements FN.Pu
         message: FN.Pusher.PusherWS.PusherMessage,
         signatureToCheck: string,
     ): Promise<boolean> {
-        let token = await this.app.createToken(
+        const token = await this.app.createToken(
             this.getDataToSignForSignature(socketId, message)
         );
 
-        let expectedSignature = this.app.key + ':' + token;
-
-        return expectedSignature === signatureToCheck;
+        return this.app.key + ':' + token === signatureToCheck;
     }
 
     getDataToSignForSignature(socketId: string, message: FN.Pusher.PusherWS.PusherMessage): string {

src/pusher/ws/pusher-connection.ts

@@ -4,7 +4,9 @@ import { Connection as BaseConnection } from '../../ws';
 export class PusherConnection extends BaseConnection implements FN.Pusher.PusherWS.PusherConnection {
     subscribedChannels: Set<string>;
     presence: Map<string, FN.Pusher.PusherWS.Presence.PresenceMember>;
-    timeout: any;
+    timeout: NodeJS.Timeout;
+    userAuthenticationTimeout: NodeJS.Timeout;
+    user: FN.JSON.Object|null;
 
     constructor(
         public id: FN.WS.ConnectionID|null,
@@ -15,6 +17,7 @@ export class PusherConnection extends BaseConnection implements FN.Pusher.Pusher
         this.id = id || this.generateSocketId();
         this.subscribedChannels = new Set();
         this.presence = new Map();
+        this.user = null;
     }
 
     protected generateSocketId(): string {
@@ -70,6 +73,7 @@ export class PusherConnection extends BaseConnection implements FN.Pusher.Pusher
     toRemote(remoteInstanceId?: string): FN.Pusher.PusherWS.PusherRemoteConnection {
         return {
             ...super.toRemote(remoteInstanceId),
+            user: this.user,
             subscribedChannels: [...this.subscribedChannels],
             presence: [...this.presence].map(([channel, member]) => ({ channel, member })),
         };

src/pusher/ws/pusher-connections.ts

@@ -8,6 +8,8 @@ import { Gossiper } from '../../gossiper';
 export class PusherConnections extends BaseConnections implements FN.Pusher.PusherWS.PusherConnections {
     readonly started: Date;
     readonly channels: Map<string, Set<string>> = new Map;
+    readonly users: Map<FN.Pusher.PusherWS.UserID, Set<FN.WS.ConnectionID>> = new Map();
+
     constructor(
         protected app: FN.Pusher.PusherApps.App,
         protected readonly gossiper: Gossiper,
@@ -18,6 +20,56 @@ export class PusherConnections extends BaseConnections implements FN.Pusher.Push
         this.started = new Date;
     }
 
+    async newConnection(conn: FN.Pusher.PusherWS.PusherConnection): Promise<void> {
+        await super.newConnection(conn);
+
+        await conn.sendJson({
+            event: 'pusher:connection_established',
+            data: JSON.stringify({
+                socket_id: conn.id,
+                activity_timeout: 30,
+            }),
+        });
+
+        if (this.app.enableUserAuthentication) {
+            conn.userAuthenticationTimeout = setTimeout(() => {
+                conn.sendError({
+                    event: 'pusher:error',
+                    data: {
+                        code: 4009,
+                        message: 'Connection not authorized within timeout.',
+                    },
+                }, 4009, 'Connection not authorized within timeout.');
+            }, this.app.userAuthenticationTimeout);
+        }
+    }
+
+    async removeConnection(conn: FN.Pusher.PusherWS.PusherConnection): Promise<void> {
+        conn.closed = true;
+
+        if (conn.timeout) {
+            clearTimeout(conn.timeout);
+        }
+
+        if (conn.userAuthenticationTimeout) {
+            clearTimeout(conn.userAuthenticationTimeout);
+        }
+
+        await this.unsubscribeFromAllChannels(conn);
+
+        if (conn.user) {
+            if (this.users.has(conn.user.id)) {
+                this.users.get(conn.user.id).delete(conn.id);
+            }
+
+            if (this.users.get(conn.user.id) && this.users.get(conn.user.id).size === 0) {
+                this.users.delete(conn.user.id);
+            }
+        }
+
+        super.removeConnection(conn);
+    }
+
     async addToChannel(conn: FN.Pusher.PusherWS.PusherConnection, channel: string): Promise<number> {
         if (!this.channels.has(channel)) {
             this.channels.set(channel, new Set);
@@ -268,6 +320,69 @@ export class PusherConnections extends BaseConnections implements FN.Pusher.Push
         // );
     }
 
+    async handleSignin(
+        conn: FN.Pusher.PusherWS.PusherConnection,
+        message: FN.Pusher.PusherWS.PusherMessage,
+    ): Promise<void> {
+        if (!conn.userAuthenticationTimeout) {
+            return;
+        }
+
+        const tokenIsValid = await this.app.signinTokenIsValid(
+            conn.id,
+            message.data.user_data,
+            message.data.auth,
+        );
+
+        if (!tokenIsValid) {
+            return conn.sendError({
+                event: 'pusher:error',
+                data: {
+                    code: 4009,
+                    message: 'Connection not authorized.',
+                },
+            }, 4009, 'Connection not authorized.');
+        }
+
+        const decodedUser = JSON.parse(message.data.user_data);
+
+        if (!decodedUser.id) {
+            return conn.sendError({
+                event: 'pusher:error',
+                data: {
+                    code: 4009,
+                    message: 'The returned user data must contain the "id" field.',
+                },
+            }, 4009, 'The returned user data must contain the "id" field.');
+        }
+
+        conn.user = {
+            ...decodedUser,
+            ...{
+                id: decodedUser.id.toString(),
+            },
+        };
+
+        if (conn.userAuthenticationTimeout) {
+            clearTimeout(conn.userAuthenticationTimeout);
+        }
+
+        if (conn.user) {
+            if (!this.users.has(conn.user.id)) {
+                this.users.set(conn.user.id, new Set());
+            }
+
+            if (!this.users.get(conn.user.id).has(conn.id)) {
+                this.users.get(conn.user.id).add(conn.id);
+            }
+        }
+
+        conn.sendJson({
+            event: 'pusher:signin_success',
+            data: message.data,
+        });
+    }
+
     async getConnections(forceLocal = false): Promise<Map<
         string,
         FN.Pusher.PusherWS.PusherConnection|FN.Pusher.PusherWS.PusherRemoteConnection
@@ -473,6 +588,34 @@ export class PusherConnections extends BaseConnections implements FN.Pusher.Push
         return this.callMethodAggregators.getChannelMembersCount(gossipResponses, options);
     }
 
+    async terminateUserConnections(userId: FN.Pusher.PusherWS.UserID, forceLocal = false): Promise<void> {
+        if (forceLocal) {
+            const connectionIds = this.users.get(userId.toString()) || new Set<string>();
+
+            for await (let connId of [...connectionIds]) {
+                this.connections.get(connId)?.close(4009, 'You got disconnected by the app.');
+            }
+
+            return;
+        }
+
+        let options: FN.Pusher.PusherGossip.GossipDataOptions = {
+            appId: this.app.id,
+            userId: userId.toString(),
+        };
+
+        this.callOthers({
+            topic: 'callMethod',
+            data: {
+                methodToCall: 'terminateUserConnections',
+                options,
+            },
+        });
+
+        // Also terminate locally.
+        this.terminateUserConnections(userId.toString(), true);
+    }
+
     async send(channel: string, data: FN.Pusher.PusherWS.SentPusherMessage, exceptingId: string|null = null, forceLocal = false): Promise<void> {
         if (forceLocal) {
             let connections = await this.getChannelConnections(channel, true) as Map<string, FN.Pusher.PusherWS.PusherConnection>;
@@ -672,6 +815,9 @@ export class PusherConnections extends BaseConnections implements FN.Pusher.Push
 
                 return localChannelMembersCount;
             },
+            terminateUserConnections: async (gossipResponses: FN.Gossip.Response[], options?: FN.Pusher.PusherGossip.GossipDataOptions) => {
+                //
+            },
             send: async (gossipResponses: FN.Gossip.Response[], options?: FN.Pusher.PusherGossip.GossipDataOptions) => {
                 //
             },
@@ -715,6 +861,17 @@ export class PusherConnections extends BaseConnections implements FN.Pusher.Push
             getChannelMembersCount: async ({ channel }: FN.Pusher.PusherGossip.GossipDataOptions) => ({
                 totalCount: await this.getChannelMembersCount(channel, true),
             }),
+            terminateUserConnections: async ({ userId, appId }: FN.Pusher.PusherGossip.GossipDataOptions) => {
+                if (appId !== this.app.id) {
+                    return;
+                }
+
+                await this.terminateUserConnections(userId, true);
+
+                return {
+                    //
+                };
+            },
             send: async ({ channel, sentPusherMessage, exceptingId }: FN.Pusher.PusherGossip.GossipDataOptions) => {
                 await this.send(
                     channel,

tests/pusher/apps/apps-registry.test.ts

@@ -84,7 +84,7 @@ class TestApp extends App {
     }
 
     async createToken(params: string): Promise<string> {
-        return Pusher.Token(this.key, this.secret).sign(params);
+        return new Pusher.Token(this.key, this.secret).sign(params);
     }
 
     async sha256(): Promise<string> {

tests/pusher/channels/public.test.ts

@@ -118,7 +118,7 @@ class TestApp extends App {
     }
 
     async createToken(params: string): Promise<string> {
-        return Pusher.Token(this.key, this.secret).sign(params);
+        return new Pusher.Token(this.key, this.secret).sign(params);
     }
 
     async sha256(): Promise<string> {