Enable authentication for QUIC transport.

QUIC transport is authenticated by transport ID which is issued by
the signaling module of portal. In the future, when QUIC transport
is allowed for signaling, other authentication method should be used. A
possible one is token, which is similar to the socket.io channel we are
using today.

Author: jianjunz

source/agent/quic/index.js

@@ -2,6 +2,18 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 
+// QUIC agent starts a QUIC transport server which listens incoming QUIC
+// connections. After a new QUIC connection is established, it waits for a
+// transport ID transmitted through signaling stream(content session ID is 0).
+// Transport ID is issued by portal. QUIC agent gets authorized transport ID
+// from |options| of publication and subscription.
+// QuicStreamPipeline is 1:1 mapped to a publication or a subscription. But the
+// QuicStream associated with QuicStreamPipeline could be a QuicStream instance
+// or |undefined|, which means QuicStream is not ready at this moment. It also
+// allows updating its associated QuicStream to a new one.
+// publish, unpublish, subscribe, unsubscribe, linkup, cutoff are required by
+// all agents. AFAIK, there is no documentation about these interfaces.
+
 'use strict';
 const Connections = require('./connections');
 const InternalConnectionFactory = require('./InternalConnectionFactory');
@@ -34,7 +46,13 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
     quicTransportServer.on('streamadded', (stream) => {
         const conn = connections.getConnection(stream.contentSessionId);
         if (conn) {
+            // TODO: verify transport ID.
             conn.connection.quicStream(stream);
+        } else {
+          log.warn(
+              'Cannot find a pipeline for QUIC stream. Content session ID: ' +
+              stream.contentSessionId);
+          stream.close();
         }
     });
 
@@ -104,6 +122,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
                 conn.connect(options);
             break;
         case 'quic':
+            quicTransportServer.addTransportId(options.transport.id);
             conn = createStreamPipeline(connectionId, 'in', options, callback);
             if (!conn) {
                 return;
@@ -155,8 +174,9 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
                     conn.connect(options);//FIXME: May FAIL here!!!!!
                 break;
             case 'quic':
+                quicTransportServer.addTransportId(options.transport.id);
                 conn = createStreamPipeline(connectionId, 'out', options, callback);
-                const stream = quicTransportServer.createSendStream('id', connectionId);
+                const stream = quicTransportServer.createSendStream(options.transport.id, connectionId);
                 conn.quicStream(stream);
                 if (!conn) {
                     return;

source/agent/quic/webtransport/quicTransportServer.js

@@ -22,27 +22,36 @@ module.exports = class QuicTransportServer extends EventEmitter {
   constructor(port) {
     super();
     this._server = new addon.QuicTransportServer(port);
-    this._connections = new Map(); // Key is user ID.
+    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._server.onconnection = (connection) => {
       this._unAuthenticatedConnections.push(connection);
       connection.onincomingstream = (stream) => {
         stream.oncontentsessionid = (id) => {
           const streamId = this._uuidBytesToString(new Uint8Array(id))
           stream.contentSessionId = streamId;
+          stream.transportId = connection.transportId;
           if (streamId === zeroUuid) {
-            // Signaling stream. Connect to portal in the future. For now, we use it for authentication.
+            // Signaling stream. Waiting for transport ID.
+          } else if (!connection.transportId) {
+            log.error(
+                'Stream ' + streamId +
+                ' added on unauthroized transport. Close connection.');
           } else {
+            log.debug(
+                'A new stream ' + streamId + ' is created on transport ' +
+                connection.transportId);
             this.emit('streamadded', stream);
           }
-        }
+        };
         stream.ondata=(data)=>{
           if (stream.contentSessionId === zeroUuid) {
-            connection.userId = data.toString('utf8');
-            this._connections.set('id', connection);
-            log.info('User ID: ' + connection.userId);
+            connection.transportId = data.toString('utf8');
+            this._connections.set(connection.transportId, connection);
+            log.info('Transport ID: ' + connection.transportId);
           }
         }
       }
@@ -51,27 +60,35 @@ module.exports = class QuicTransportServer extends EventEmitter {
 
   start() {
     this._server.start();
+    // TODO: Return error if server is not started, e.g.: port is not available.
+  }
+
+  addTransportId(id) {
+    if (!(id in this._assignedTransportIds)) {
+      this._assignedTransportIds.push(id);
+    }
   }
 
-  createSendStream(userId, contentSessionId) {
-    userId = 'id';
-    if (!this._connections.has(userId)) {
-      log.error('No QUIC transport for '+userId);
+  // Create a send stream for specfied QuicTransport. If specified QuicTransport
+  // doesn't exist, no stream will be created.
+  createSendStream(transportId, contentSessionId) {
+    log.debug(
+        'Create send stream ' + contentSessionId + ' on transport ' +
+        transportId);
+    if (!this._connections.has(transportId)) {
+      // TODO: Waiting for transport to be created, and create stream for it.
+      // It's a common case that subscribe request is received by QUIC agent
+      // before client creates QuicTransport.
+      log.error('No QUIC transport for ' + transportId);
       return;
     }
-    const stream = this._connections.get(userId).createBidirectionalStream();
-    log.info('Stream: ' + JSON.stringify(stream));
+    const stream = this._connections.get(transportId).createBidirectionalStream();
     const uuidBytes=this._uuidStringToUint8Array(contentSessionId);
-    log.info('Data: '+uuidBytes);
     stream.write(uuidBytes, uuidBytes.length);
     this._streams.set(contentSessionId, stream);
     return stream;
   }
 
-  _authenticate(){
-    // TODO: Send token to nuve for authentication.
-  }
-
   _uuidBytesToString(uuidBytes) {
     if (uuidBytes.length != 16) {
       log.error('Invalid UUID.');

source/portal/requestDataValidator.js

@@ -71,7 +71,8 @@ const PublicationRequest = {
         'transport': {
           type: 'object',
           properties: {
-            'type': {type: 'string'}
+            'type': {type: 'string'},
+            'id': {type: 'string'},
           },
           additionalProperties: false,
         }
@@ -180,7 +181,8 @@ const SubscriptionRequest = {
         'transport': {
           type: 'object',
           properties: {
-            'type': {type: 'string'}
+            'type': {type: 'string'},
+            'id': {type: 'string'},
           },
         },
       },

source/portal/v11Client.js

@@ -107,6 +107,10 @@ var V11Client = function(clientId, sigConnection, portal) {
       });
   };
 
+  const uuidWithoutDash = function() {
+    return uuid().replace(/-/g, '');
+  };
+
   const listenAt = (socket) => {
     socket.on('text', function(textReq, callback) {
       if(!that.inRoom){
@@ -126,17 +130,22 @@ var V11Client = function(clientId, sigConnection, portal) {
         return safeCall(callback, 'error', 'Illegal request');
       }
 
-      var stream_id = uuid().replace(/-/g,'');
+      var stream_id = uuidWithoutDash();
+      let transport_id;
       return validatePubReq(pubReq)
         .then((req) => {
           if (pubReq.transport && pubReq.transport.type == 'quic') {
             req.type = 'quic';
+            if (!req.transport.id) {
+                req.transport.id = uuidWithoutDash();
+            }
+            transport_id = req.transport.id;
           } else {
             req.type = 'webrtc'; //FIXME: For backend compatibility with v3.4 clients.
           }
           return portal.publish(clientId, stream_id, req);
         }).then((result) => {
-          safeCall(callback, 'ok', {id: stream_id});
+          safeCall(callback, 'ok', {id: stream_id, transportId: transport_id});
         }).catch(onError('publish', callback));
     });
 
@@ -172,16 +181,21 @@ var V11Client = function(clientId, sigConnection, portal) {
       }
 
       var subscription_id = uuid().replace(/-/g,'');
+      let transport_id;
       return validateSubReq(subReq)
         .then((req) => {
           if (req.transport && req.transport.type == 'quic') {
             req.type = 'quic';
+            if (!req.transport.id) {
+              req.transport.id = uuidWithoutDash();
+            }
+            transport_id = req.transport.id;
           } else {
             req.type = 'webrtc';//FIXME: For backend compatibility with v3.4 clients.
           }
           return portal.subscribe(clientId, subscription_id, req);
         }).then((result) => {
-          safeCall(callback, 'ok', {id: subscription_id});
+          safeCall(callback, 'ok', {id: subscription_id, transportId: transport_id});
         }).catch(onError('subscribe', callback));
     });