adds support for kick/block/unblock

Author: InfiniteLee

.eslintrc.js

@@ -170,3 +170,65 @@ module.exports =
 		'yoda': 2
 	}
 };
+
+switch (process.env.MEDIASOUP_NODE_LANGUAGE)
+{
+	case 'typescript':
+	{
+		eslintConfig.parser = '@typescript-eslint/parser';
+		eslintConfig.plugins =
+		[
+			...eslintConfig.plugins,
+			'@typescript-eslint'
+		];
+		eslintConfig.extends =
+		[
+			'eslint:recommended',
+			'plugin:@typescript-eslint/eslint-recommended',
+			'plugin:@typescript-eslint/recommended'
+		];
+		eslintConfig.rules =
+		{
+			...eslintConfig.rules,
+			'no-unused-vars'                            : 0,
+			'@typescript-eslint/ban-ts-ignore'          : 0,
+			'@typescript-eslint/member-delimiter-style' : [ 2,
+				{
+					multiline  : { delimiter: 'semi', requireLast: true },
+					singleline : { delimiter: 'semi', requireLast: false }
+				}
+			],
+			'@typescript-eslint/no-explicit-any' : 0,
+			'@typescript-eslint/no-unused-vars'  : [ 2,
+				{
+					vars               : 'all',
+					args               : 'after-used',
+					ignoreRestSiblings : false
+				}
+			],
+			'@typescript-eslint/no-use-before-define' : 0,
+			'@typescript-eslint/no-empty-function'    : 0
+		};
+
+		break;
+	}
+
+	case 'javascript':
+	{
+		eslintConfig.env['jest/globals'] = true;
+		eslintConfig.plugins =
+		[
+			...eslintConfig.plugins,
+			'jest'
+		];
+
+		break;
+	}
+
+	default:
+	{
+		throw new TypeError('wrong/missing MEDIASOUP_NODE_LANGUAGE env');
+	}
+}
+
+module.exports = eslintConfig;

config.js

@@ -133,7 +133,8 @@ const config =
 			// Additional options that are not part of WebRtcTransportOptions.
 			maxIncomingBitrate              : 1500000
 		}
-	}
+	},
+	authKey: process.env.AUTH_KEY || `${__dirname}/certs/perms.pub.pem`
 };
 
 if (process.env.MEDIASOUP_ANNOUNCED_IP) 

index.js

@@ -25,6 +25,8 @@ const Logger = require('./lib/Logger');
 const Room = require('./lib/Room');
 const interactiveServer = require('./lib/interactiveServer');
 const interactiveClient = require('./lib/interactiveClient');
+const util = require('util');
+const readFile = util.promisify(fs.readFile);
 
 const logger = new Logger();
 const queue = new AwaitQueue();
@@ -37,6 +39,8 @@ let protooWebSocketServer;
 const mediasoupWorkers = [];
 let nextMediasoupWorkerIdx = 0;
 
+let authKey;
+
 run();
 
 async function run()
@@ -49,8 +53,14 @@ async function run()
 	await runMediasoupWorkers();
 	await createExpressApp();
 	await runHttpsServer();
+	try {
+		authKey = await readFile(config.authKey, 'utf8');
+	} catch (error) {
+		logger.error("authKey not set; jwt verification will not work.", error);
+	}
 	await runProtooWebSocketServer();
 
+
 	// Log rooms status every X seconds.
 	setInterval(() =>
 	{
@@ -418,7 +428,7 @@ async function getOrCreateRoom({ roomId })
 
 		const mediasoupWorker = getMediasoupWorker();
 
-		room = await Room.create({ mediasoupWorker, roomId });
+		room = await Room.create({ mediasoupWorker, roomId, authKey });
 
 		rooms.set(roomId, room);
 		room.on('close', () => rooms.delete(roomId));

lib/Room.js

@@ -4,6 +4,7 @@ const throttle = require('@sitespeed.io/throttle');
 const Logger = require('./Logger');
 const config = require('../config');
 const Bot = require('./Bot');
+const jwt = require('jsonwebtoken');
 
 const logger = new Logger('Room');
 
@@ -25,7 +26,7 @@ class Room extends EventEmitter
 	 *   mediasoup Router must be created.
 	 * @param {String} roomId - Id of the Room instance.
 	 */
-	static async create({ mediasoupWorker, roomId })
+	static async create({ mediasoupWorker, roomId, authKey })
 	{
 		logger.info('create() [roomId:%s]', roomId);
 
@@ -54,11 +55,12 @@ class Room extends EventEmitter
 				protooRoom,
 				mediasoupRouter,
 				audioLevelObserver,
-				bot
+				bot,
+				authKey
 			});
 	}
 
-	constructor({ roomId, protooRoom, mediasoupRouter, audioLevelObserver, bot })
+	constructor({ roomId, protooRoom, mediasoupRouter, audioLevelObserver, bot, authKey })
 	{
 		super();
 		this.setMaxListeners(Infinity);
@@ -105,6 +107,8 @@ class Room extends EventEmitter
 		// @type {Boolean}
 		this._networkThrottled = false;
 
+		this._authKey = authKey;
+
 		// Handle audioLevelObserver.
 		this._handleAudioLevelObserver();
 
@@ -202,6 +206,8 @@ class Room extends EventEmitter
 		peer.data.consumers = new Map();
 		peer.data.dataProducers = new Map();
 		peer.data.dataConsumers = new Map();
+		peer.data.peerToConsumers = new Map();
+		peer.data.blockedPeers = new Set();
 
 		peer.on('request', (request, accept, reject) =>
 		{
@@ -705,7 +711,8 @@ class Room extends EventEmitter
 					displayName,
 					device,
 					rtpCapabilities,
-					sctpCapabilities
+					sctpCapabilities,
+					token
 				} = request.data;
 
 				// Store client data into the protoo Peer data object.
@@ -714,6 +721,7 @@ class Room extends EventEmitter
 				peer.data.device = device;
 				peer.data.rtpCapabilities = rtpCapabilities;
 				peer.data.sctpCapabilities = sctpCapabilities;
+				peer.data.token = token;
 
 				// Tell the new Peer about already joined Peers.
 				// And also create Consumers for existing Producers.
@@ -1387,6 +1395,71 @@ class Room extends EventEmitter
 				break;
 			}
 
+			case 'kick':
+			{
+				jwt.verify(peer.data.token, this._authKey, { algorithms: ['RS512'] }, (err, decoded) => {
+					if (err) {
+						reject(500, err);
+					} else {
+						if (decoded.kick_users) {
+							const consumerId = request.data.user_id;
+							if (this._protooRoom.hasPeer(consumerId)) {
+								this._protooRoom.getPeer(consumerId).close();
+							}
+							accept();
+						} else {
+							reject(401);
+						}
+					}
+				});
+				break;
+			}
+			
+			case 'block':
+			{
+				const localPeer = peer;
+				const remotePeerId = request.data.whom;
+
+				const localConsumerId = localPeer.data.peerToConsumers.get(remotePeerId);
+				if (localPeer.data.consumers.has(localConsumerId)) {
+					localPeer.data.blockedPeers.add(remotePeerId);
+					localPeer.data.consumers.get(localConsumerId).pause();
+					
+					if (this._protooRoom.hasPeer(remotePeerId)) {
+						const remotePeer = this._protooRoom.getPeer(remotePeerId)
+						const remoteConsumerId = remotePeer.data.peerToConsumers.get(localPeer.id);
+						if(remotePeer.data.consumers.has(remoteConsumerId)) {
+							remotePeer.data.consumers.get(remoteConsumerId).pause();
+							remotePeer.notify('peerBlocked', { peerId: localPeer.id }).catch(() => {});
+						}
+					}
+				}	
+				accept();
+				break;
+			}
+
+			case 'unblock':
+			{
+				const localPeer = peer;
+				const remotePeerId = request.data.whom;
+				
+				localPeer.data.blockedPeers.delete(remotePeerId);
+
+				if (this._protooRoom.hasPeer(remotePeerId)) {
+					const remotePeer = this._protooRoom.getPeer(remotePeerId);
+					if (!remotePeer.data.blockedPeers.has(localPeer.id) && 
+						localPeer.data.consumers.has(localPeer.data.peerToConsumers.get(remotePeerId)) && 
+						remotePeer.data.consumers.has(remotePeer.data.peerToConsumers.get(localPeer.id))) {
+						localPeer.data.consumers.get(localPeer.data.peerToConsumers.get(remotePeerId)).resume();
+						remotePeer.data.consumers.get(remotePeer.data.peerToConsumers.get(localPeer.id)).resume();
+						localPeer.notify('peerUnblocked', { peerId: remotePeer.id }).catch(() => {});
+						remotePeer.notify('peerUnblocked', { peerId: localPeer.id }).catch(() => {});
+					}
+				}
+				accept();
+				break;
+			}
+
 			default:
 			{
 				logger.error('unknown request.method "%s"', request.method);
@@ -1471,17 +1544,21 @@ class Room extends EventEmitter
 		// Store the Consumer into the protoo consumerPeer data Object.
 		consumerPeer.data.consumers.set(consumer.id, consumer);
 
+		consumerPeer.data.peerToConsumers.set(producerPeer.id, consumer.id);
+
 		// Set Consumer events.
 		consumer.on('transportclose', () =>
 		{
 			// Remove from its map.
 			consumerPeer.data.consumers.delete(consumer.id);
+			consumerPeer.data.peerToConsumers.delete(producerPeer.id);
 		});
 
 		consumer.on('producerclose', () =>
 		{
 			// Remove from its map.
 			consumerPeer.data.consumers.delete(consumer.id);
+			consumerPeer.data.peerToConsumers.delete(producerPeer.id);
 
 			consumerPeer.notify('consumerClosed', { consumerId: consumer.id })
 				.catch(() => {});

package.json

@@ -17,6 +17,7 @@
     "debug": "^4.1.1",
     "express": "^4.17.1",
     "heapdump": "^0.3.15",
+    "jsonwebtoken": "^8.5.1",
     "mediasoup": "github:versatica/mediasoup#v3",
     "pidusage": "^2.0.18",
     "protoo-server": "^4.0.4",