Authenticate WebTransport connections with tokens.

Tokens are issued by portal after joining. QUIC agent asks portal for
authentication.

Author: jianjunz

source/agent/conference/conference.js

@@ -1843,6 +1843,15 @@ var Conference = function (rpcClient, selfRpcId) {
     callback('callback', result);
   };
 
+  that.getPortal = function(participantId, callback) {
+      log.debug('Get participant ' + participantId);
+      if (!participants[participantId]) {
+          callback('callback', 'error', 'Invalid participant ID.');
+          return;
+      }
+      callback('callback', participants[participantId].getPortal());
+  };
+
   //FIXME: Should handle updates other than authorities as well.
   that.controlParticipant = function(participantId, authorities, callback) {
     log.debug('controlParticipant', participantId, 'authorities:', authorities);
@@ -2674,7 +2683,10 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
     controlSipCall: conference.controlSipCall,
     endSipCall: conference.endSipCall,
     drawText: conference.drawText,
-    destroy: conference.destroy
+    destroy: conference.destroy,
+
+    // RPC from QUIC nodes.
+    getPortal: conference.getPortal
   };
 
   that.onFaultDetected = conference.onFaultDetected;

source/agent/quic/dist.json

@@ -18,8 +18,6 @@
           "../../common/clusterWorker.js",
           "../../common/loadCollector.js",
           "../../common/logger.js",
-          "../../common/makeRPC.js",
-          "../../common/rpcChannel.js",
           "../../../scripts/release/initcert.js",
           "../../../scripts/release/initauth.js",
           "../../../scripts/detectOS.sh"
@@ -28,7 +26,9 @@
           "quic": [
               "index.js",
               "../connections.js",
-              "../InternalConnectionFactory.js"
+              "../InternalConnectionFactory.js",
+              "../../common/makeRPC.js",
+              "../../common/rpcChannel.js"
           ],
           "quic/webtransport": [
               "webtransport/quicTransportServer.js",

source/agent/quic/index.js

@@ -29,6 +29,7 @@ const path = require('path');
 log.info('QUIC transport node.')
 
 module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
+    const rpcChannel = require('./rpcChannel')(rpcClient);
     const that = {
       agentID: parentRpcId,
       clusterIP: clusterWorkerIP
@@ -40,23 +41,60 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
     const outgoingStreamPipelines =
         new Map();  // Key is subscription ID, value is stream pipeline.
     let quicTransportServer;
+    const clusterName = global && global.config && global.config.cluster ?
+        global.config.cluster.name :
+        undefined;
+
+    const getRpcPortal = async (roomId, participantId) => {
+      const controllerAgent =
+          await rpcChannel.makeRPC(clusterName, 'schedule', [
+            'conference', roomId, 'preference' /*TODO: specify preference*/,
+            30 * 1000
+          ]);
+      let controller = null;
+      let portal = null;
+      const retry = 5;
+      const sleep = require('util').promisify(setTimeout);
+      for (let i = 0; i < retry; i++) {
+        try {
+          controller = await rpcChannel.makeRPC(
+              controllerAgent.id, 'getNode', [{room: roomId, task: roomId}]);
+          if (controller) {
+            break;
+          }
+        } catch (error) {
+          sleep(1000);
+        }
+      }
+      portal =
+          await rpcChannel.makeRPC(controller, 'getPortal', [participantId])
+      return portal;
+    };
 
     const notifyStatus = (controller, sessionId, direction, status) => {
-        rpcClient.remoteCast(controller, 'onSessionProgress', [sessionId, direction, status]);
+      rpcClient.remoteCast(
+          controller, 'onSessionProgress', [sessionId, direction, status]);
+    };
+
+    const validateToken = async (token) => {
+      if (!token || !token.roomId) {
+        throw new TypeError('Invalid token.');
+      }
+      const portal = await getRpcPortal(token.roomId, token.participantId);
+      return rpcChannel.makeRPC(
+          portal, 'validateAndDeleteWebTransportToken', [token])
     };
 
     const keystore = path.resolve(path.dirname(global.config.quic.keystorePath), cipher.kstore);
-    log.info('before unlock');
     cipher.unlock(cipher.k, keystore, (error, password) => {
-      log.info('unlocked.');
       if (error) {
         log.error('Failed to read certificate and key.');
         return;
       }
       log.info('path is '+path.resolve(global.config.quic.keystorePath));
       quicTransportServer = new QuicTransportServer(
           addon, global.config.quic.port, path.resolve(global.config.quic.keystorePath),
-          password);
+          password, validateToken);
       quicTransportServer.start();
       quicTransportServer.on('streamadded', (stream) => {
         const conn = connections.getConnection(stream.contentSessionId);

source/agent/quic/webtransport/quicTransportServer.js

@@ -23,14 +23,15 @@ const zeroUuid = '00000000000000000000000000000000';
 const authenticationTimeout = 10;
 
 module.exports = class QuicTransportServer extends EventEmitter {
-  constructor(addon, port, pfxPath, password) {
+  constructor(addon, port, pfxPath, password, validateTokenCallback) {
     super();
     this._server = new addon.QuicTransportServer(port, pfxPath, password);
     this._connections = new Map(); // Key is transport ID.
     this._streams = new Map(); // Key is content session ID.
     this._unAuthenticatedConnections = []; // When it's authenticated, it will be moved to this.connections.
     this._unAssociatedStreams = []; // No content session ID assgined to them.
     this._assignedTransportIds = []; // Transport channels assigned to this server.
+    this._validateTokenCallback = validateTokenCallback;
     this._server.onconnection = (connection) => {
       this._unAuthenticatedConnections.push(connection);
       setTimeout(() => {
@@ -61,16 +62,68 @@ module.exports = class QuicTransportServer extends EventEmitter {
         };
         stream.ondata = (data) => {
           if (stream.contentSessionId === zeroUuid) {
-            const transportId = data.toString('utf8');
-            log.info('Received new transport ID: '+transportId);
-            if (!this._assignedTransportIds.includes(transportId)) {
-              log.info(JSON.stringify(connection));
-              connection.close();
+            // Please refer
+            // https://github.com/open-webrtc-toolkit/owt-server/blob/20e8aad216cc446095f63c409339c34c7ee770ee/doc/design/quic-transport-payload-format.md#signaling-session
+            // for the format of data received on this stream.
+
+            // Size of data received, but haven't been read.
+            let unreadSize = data.length;
+            let nextReadSize = stream.frameSize ? stream.frameSize : 0;
+            while (unreadSize != 0) {
+              if (nextReadSize == 0) {  // Starts a new frame.
+                // 32 bit indicates the length of next frame.
+                const lengthSize = 4;
+                if (data.length >= lengthSize) {
+                  for (let i = 0; i < lengthSize; i++) {
+                    nextReadSize += (data[i] << 8 * i);
+                  }
+                  unreadSize -= lengthSize;
+                } else {
+                  log.error('Not implemented.');
+                  return;
+                }
+              } else {
+                nextReadSize = stream.frameSize - stream.unreadData.length;
+              }
+              const startIndex = data.length - unreadSize;
+              if (unreadSize >= nextReadSize) {
+                let frame = data.slice(startIndex, startIndex + nextReadSize);
+                if (stream.unreadData) {
+                  frame = stream.unreadData + frame;
+                  stream.unreadData = null;
+                  stream.frameSize = 0;
+                }
+                const message = frame.toString('utf8');
+                unreadSize -= nextReadSize;
+                // Currently, Signaling over WebTransport is not supported, so
+                // only one message (WebTransport token) will be sent on this
+                // stream. We process the message and discard all other
+                // messages.
+                let token;
+                try {
+                  token = JSON.parse(Buffer.from(message, 'base64').toString());
+                } catch (error) {
+                  log.error('Invalid token.');
+                  connection.close();
+                  return;
+                }
+                this._validateTokenCallback(token).then(result => {
+                  if (result !== 'ok') {
+                    log.error('Authentication failed.');
+                    connection.close();
+                    return;
+                  }
+                  connection.transportId = token.transportId;
+                  stream.transportId = token.transportId;
+                  this._connections.set(connection.transportId, connection);
+                });
+                return;
+              } else {  // Store the data.
+                stream.unreadData = data.slice(startIndex);
+                stream.frameSize = nextReadSize;
+              }
             }
-            connection.transportId = transportId;
-            stream.transportId = transportId;
-            this._connections.set(connection.transportId, connection);
-            log.info('New connection for transport ID: ' + transportId);
+            return;
           }
         }
       }
@@ -132,4 +185,9 @@ module.exports = class QuicTransportServer extends EventEmitter {
     }
     return uuidArray;
   }
+
+  _validateToken(tokenString) {
+    const token = JSON.parse(Buffer.from(tokenString, 'base64').toString());
+    return this._validateTokenCallback(token);
+  }
 };

source/portal/index.js

@@ -233,11 +233,11 @@ var rpcPublic = {
     socketio_server && socketio_server.notify(participantId, event, data).catch(notifyFail);
     callback('callback', 'ok');
   },
-  validateWebTransportToken: (token, callback) => {
-    if(portal.validateWebTransportToken(token)) {
+  validateAndDeleteWebTransportToken: (token, callback) => {
+    if(portal.validateAndDeleteWebTransportToken(token)) {
       callback('callback','ok');
     } else {
-      callback('callback', 'error');
+      callback('callback', 'error', 'Invalid token for WebTransport.');
     }
   }
 };

source/portal/portal.js

@@ -26,36 +26,46 @@ var Portal = function(spec, rpcReq) {
    */
   var participants = {};
 
-  // Key is token, value is participant ID. An ID is only valid when the participant is online.
+  // Key is participantId, value is token ID.
   const webTransportIds = new Map();
   const calculateSignatureForWebTransportToken = (token) => {
-      const toSign = vsprintf('%s,%s,%s,%s', [
+      const toSign = vsprintf('%s,%s,%s,%s,%s', [
           token.tokenId,
           token.transportId,
           token.participantId,
+          token.roomId,
           token.issueTime
       ]);
       const signed = crypto.createHmac('sha256', token_key).update(toSign).digest('hex');
       return (Buffer.from(signed)).toString('base64');
   };
-  const generateWebTransportToken = (participantId) => {
+  const generateWebTransportToken = (participantId, roomId) => {
       const now = Date.now();
       const token = {
           tokenId : uuid().replace(/-/g, ''),
           transportId: uuid().replace(/-/g, ''),
           participantId : participantId,
+          roomId: roomId,
           issueTime : now,
       };
       token.signature = calculateSignatureForWebTransportToken(token);
+      webTransportIds.set(participantId, token.tokenId);
       return token;
   };
 
-  that.validateWebTransportToken = (token) => {
+  that.validateAndDeleteWebTransportToken = (token) => {
       // |participants| is better to be a map.
       if (!participants.hasOwnProperty(token.participantId)) {
           return false;
       }
-      return calculateSignatureForWebTransportToken(token) == token.signature;
+      if (!webTransportIds.has(token.participantId) || webTransportIds.get(token.participantId) !== token.tokenId) {
+          return false;
+      }
+      if (calculateSignatureForWebTransportToken(token) !== token.signature) {
+          return false;
+      }
+      webTransportIds.delete(token.participantId);
+      return true;
   };
 
   that.updateTokenKey = function(tokenKey) {
@@ -114,7 +124,7 @@ var Portal = function(spec, rpcReq) {
 
         let webTransportToken = undefined;
         if (token.webTransportUrl) {
-            webTransportToken = (Buffer.from(JSON.stringify(generateWebTransportToken(participantId)))).toString('base64');
+            webTransportToken = (Buffer.from(JSON.stringify(generateWebTransportToken(participantId, room_id)))).toString('base64');
         }
 
         return {