Feature/siprec server (#140)

* initial support for siprec/agent assist

* call siprec middleware

* logger fix

* remove verbs that are not valid in a siprec call session

app.js

@@ -28,6 +28,7 @@ installSrfLocals(srf, logger);
 const {
   initLocals,
   createRootSpan,
+  handleSipRec,
   getAccountDetails,
   normalizeNumbers,
   retrieveApplication,
@@ -46,6 +47,7 @@ Object.assign(app.locals, {
 const httpRoutes = require('./lib/http-routes');
 
 const InboundCallSession = require('./lib/session/inbound-call-session');
+const SipRecCallSession = require('./lib/session/siprec-call-session');
 
 if (process.env.DRACHTIO_HOST) {
   srf.connect({host: process.env.DRACHTIO_HOST, port: process.env.DRACHTIO_PORT, secret: process.env.DRACHTIO_SECRET });
@@ -68,6 +70,7 @@ if (process.env.NODE_ENV === 'test') {
 srf.use('invite', [
   initLocals,
   createRootSpan,
+  handleSipRec,
   getAccountDetails,
   normalizeNumbers,
   retrieveApplication,
@@ -75,7 +78,8 @@ srf.use('invite', [
 ]);
 
 srf.invite((req, res) => {
-  const session = new InboundCallSession(req, res);
+  const isSipRec = !!req.locals.siprec;
+  const session = isSipRec ? new SipRecCallSession(req, res) : new InboundCallSession(req, res);
   session.exec();
 });
 

lib/middleware.js

@@ -1,5 +1,6 @@
 const { v4: uuidv4 } = require('uuid');
-const {CallDirection} = require('./utils/constants');
+const {CallDirection, TaskName} = require('./utils/constants');
+const {parseSiprecPayload} = require('./utils/siprec-utils');
 const CallInfo = require('./session/call-info');
 const HttpRequestor = require('./utils/http-requestor');
 const WsRequestor = require('./utils/ws-requestor');
@@ -73,6 +74,35 @@ module.exports = function(srf, logger) {
     next();
   }
 
+  const handleSipRec = async(req, res, next) => {
+    if (Array.isArray(req.payload) && req.payload.length > 1) {
+      const {callId, logger} = req.locals;
+      logger.debug({payload: req.payload}, 'handling siprec call');
+
+      try {
+        const sdp = req.payload
+          .find((p) => p.type === 'application/sdp')
+          .content;
+        const {sdp1, sdp2, ...metadata} = await parseSiprecPayload(req, logger);
+        req.locals.calledNumber = metadata.caller.number;
+        req.locals.callingNumber = metadata.callee.number;
+        req.locals = {
+          ...req.locals,
+          siprec: {
+            metadata,
+            sdp1,
+            sdp2
+          }
+        };
+        logger.info({callId, metadata, sdp}, 'successfully parsed SIPREC payload');
+      } catch (err) {
+        logger.info({callId}, 'Error parsing multipart payload');
+        return res.send(503);
+      }
+    }
+    next();
+  };
+
   /**
    * retrieve account information for the incoming call
    */
@@ -101,7 +131,10 @@ module.exports = function(srf, logger) {
    * Within the system, we deal with E.164 numbers _without_ the leading '+
    */
   function normalizeNumbers(req, res, next) {
-    const logger = req.locals.logger;
+    const {logger, siprec} = req.locals;
+
+    if (siprec) return next();
+
     Object.assign(req.locals, {
       calledNumber: req.calledNumber,
       callingNumber: req.callingNumber
@@ -225,7 +258,7 @@ module.exports = function(srf, logger) {
    */
   async function invokeWebCallback(req, res, next) {
     const logger = req.locals.logger;
-    const {rootSpan, application:app} = req.locals;
+    const {rootSpan, siprec, application:app} = req.locals;
     let span;
     try {
       if (app.tasks) {
@@ -261,6 +294,19 @@ module.exports = function(srf, logger) {
       });
       span.end();
       if (0 === app.tasks.length) throw new Error('no application provided');
+
+      if (siprec) {
+        /* only transcribe and/or listen allowed on an incoming siprec call */
+        const tasks = app.tasks.filter((t) => [TaskName.Config, TaskName.Listen, TaskName.Transcribe].includes(t.name));
+        if (0 === tasks.length) {
+          logger.info({tasks: app.tasks}, 'only config, transcribe and/or listen allowed on an incoming siprec call');
+          throw new Error('invalid verbs for incoming siprec call');
+        }
+        if (tasks.length < app.tasks.length) {
+          logger.info('removing verbs that are not allowed for incoming siprec call');
+          app.tasks = tasks;
+        }
+      }
       next();
     } catch (err) {
       span?.setAttributes({webhookStatus: err.statusCode});
@@ -274,6 +320,7 @@ module.exports = function(srf, logger) {
   return {
     initLocals,
     createRootSpan,
+    handleSipRec,
     getAccountDetails,
     normalizeNumbers,
     retrieveApplication,

lib/session/call-session.js

@@ -228,6 +228,13 @@ class CallSession extends Emitter {
     return this.constructor.name === 'ConfirmCallSession';
   }
 
+  /**
+   * returns true if this session is a SipRecCallSession
+   */
+  get isSipRecCallSession() {
+    return this.constructor.name === 'SipRecCallSession';
+  }
+
   /**
    * returns true if this session is a SmsCallSession
    */
@@ -908,6 +915,22 @@ class CallSession extends Emitter {
       this.logger.debug('CallSession:replaceApplication - ignoring because call is gone');
       return;
     }
+
+    if (this.isSipRecCallSession) {
+      const pruned = tasks.filter((t) =>
+        [TaskName.Config, TaskName.Listen, TaskName.Transcribe].includes(t.name)
+      );
+      if (0 === pruned.length) {
+        this.logger.info({tasks},
+          'CallSession:replaceApplication - only config, transcribe and/or listen allowed on an incoming siprec call');
+        return;
+      }
+      if (pruned.length < tasks.length) {
+        this.logger.info(
+          'CallSession:replaceApplication - removing verbs that are not allowed for incoming siprec call');
+        tasks = pruned;
+      }
+    }
     this.tasks = tasks;
     this.taskIdx = 0;
     this.stackIdx++;

lib/session/siprec-call-session.js

@@ -0,0 +1,56 @@
+const InboundCallSession = require('./inbound-call-session');
+const {createSipRecPayload} = require('../utils/siprec-utils');
+/**
+ * @classdesc Subclass of InboundCallSession.  This represents a CallSession that is
+ * established for an inbound SIPREC call.
+ * @extends InboundCallSession
+ */
+class SipRecCallSession extends InboundCallSession {
+  constructor(req, res) {
+    super(req, res);
+
+    const {sdp1, sdp2, metadata} = req.locals.siprec;
+    this.sdp1 = sdp1;
+    this.sdp2 = sdp2;
+    this.metadata = metadata;
+
+    setImmediate(this._answerSipRecCall.bind(this));
+  }
+
+  async _answerSipRecCall() {
+    try {
+      this.ms = this.getMS();
+      this.ep = await this.ms.createEndpoint({remoteSdp: this.sdp1});
+      this.ep2 = await this.ms.createEndpoint({remoteSdp: this.sdp2});
+      await this.ep.bridge(this.ep2);
+      const combinedSdp = await createSipRecPayload(this.ep.local.sdp, this.ep2.local.sdp, this.logger);
+      this.logger.debug({
+        sdp1: this.sdp1,
+        sdp2: this.sdp2,
+        combinedSdp
+      }, 'SipRecCallSession:_answerSipRecCall - created SIPREC payload');
+      this.dlg = await this.srf.createUAS(this.req, this.res, {
+        headers: {
+          'X-Trace-ID': this.req.locals.traceId,
+          'X-Call-Sid': this.req.locals.callSid
+        },
+        localSdp: combinedSdp
+      });
+    } catch (err) {
+      this.logger.error({err}, 'SipRecCallSession:_answerSipRecCall error:');
+      if (this.res && !this.res.finalResponseSent) this.res.send(500);
+      this._callReleased();
+    }
+  }
+
+  _callReleased() {
+    /* release that second endpoint we created, then call superclass implementation */
+    if (this.ep2?.connected) {
+      this.ep2.destroy();
+      this.ep2 = null;
+    }
+    super._callReleased();
+  }
+}
+
+module.exports = SipRecCallSession;

lib/utils/siprec-utils.js

@@ -0,0 +1,248 @@
+const xmlParser = require('xml2js').parseString;
+const { v4: uuidv4 } = require('uuid');
+const parseUri = require('drachtio-srf').parseUri;
+const transform = require('sdp-transform');
+const debug = require('debug')('jambonz:feature-server');
+
+const parseCallData = (prefix, obj) => {
+  const ret = {};
+  const group = obj[`${prefix}group`];
+  if (group) {
+    const key = Object.keys(group[0]).find((k) => /:?callData$/.test(k));
+    //const o = _.find(group[0], (value, key) => /:?callData$/.test(key));
+    if (key) {
+      //const callData = o[0];
+      const callData = group[0][key];
+      for (const key of Object.keys(callData)) {
+        if (['fromhdr', 'tohdr', 'callid'].includes(key)) ret[key] = callData[key][0];
+      }
+    }
+  }
+  debug('parseCallData', prefix, obj, ret);
+  return ret;
+};
+
+/**
+ * parse a SIPREC multiparty body
+ * @param  {object} opts - options
+ * @return {Promise}
+ */
+const parseSiprecPayload = (req, logger) => {
+  const opts = {};
+  return new Promise((resolve, reject) => {
+    let sdp, meta ;
+    for (let i = 0; i < req.payload.length; i++) {
+      switch (req.payload[i].type) {
+        case 'application/sdp':
+          sdp = req.payload[i].content ;
+          break ;
+
+        case 'application/rs-metadata+xml':
+        case 'application/rs-metadata':
+          meta = opts.xml = req.payload[i].content ;
+          break ;
+
+        default:
+          break ;
+      }
+    }
+
+    if (!sdp || !meta) {
+      logger.info({payload: req.payload}, 'invalid SIPREC payload');
+      return reject(new Error('expected multipart SIPREC body'));
+    }
+
+    xmlParser(meta, (err, result) => {
+      if (err) { throw err; }
+
+      opts.recordingData = result ;
+      opts.sessionId = uuidv4() ;
+
+      const arr = /^([^]+)(m=[^]+?)(m=[^]+?)$/.exec(sdp) ;
+      opts.sdp1 = `${arr[1]}${arr[2]}` ;
+      opts.sdp2 = `${arr[1]}${arr[3]}\r\n` ;
+
+      try {
+        if (typeof result === 'object' && Object.keys(result).length === 1) {
+          const key = Object.keys(result)[0] ;
+          const arr = /^(.*:)recording/.exec(key) ;
+          const prefix = !arr ? '' : (arr[1]) ;
+          const obj = opts.recordingData[`${prefix}recording`];
+
+          // 1. collect participant data
+          const participants = {} ;
+          obj[`${prefix}participant`].forEach((p) => {
+            const partDetails = {} ;
+            participants[p.$.participant_id] = partDetails;
+            if ((`${prefix}nameID` in p) && Array.isArray(p[`${prefix}nameID`])) {
+              partDetails.aor = p[`${prefix}nameID`][0].$.aor;
+              if ('name' in p[`${prefix}nameID`][0] && Array.isArray(p[`${prefix}nameID`][0].name)) {
+                const name = p[`${prefix}nameID`][0].name[0];
+                if (typeof name === 'string') partDetails.name = name ;
+                else if (typeof name === 'object') partDetails.name = name._ ;
+              }
+            }
+          });
+
+          // 2. find the associated streams for each participant
+          if (`${prefix}participantstreamassoc` in obj) {
+            obj[`${prefix}participantstreamassoc`].forEach((ps) => {
+              const part = participants[ps.$.participant_id];
+              if (part) {
+                part.send = ps[`${prefix}send`][0];
+                part.recv = ps[`${prefix}recv`][0];
+              }
+            });
+          }
+
+          // 3. Retrieve stream data
+          opts.caller = {} ;
+          opts.callee = {} ;
+          obj[`${prefix}stream`].forEach((s) => {
+            const streamId = s.$.stream_id;
+            let sender;
+            for (const [k, v] of Object.entries(participants)) {
+              if (v.send === streamId) {
+                sender = k;
+                break;
+              }
+            }
+            //const sender = _.find(participants, { 'send': streamId});
+
+            if (!sender) return;
+
+            sender.label = s[`${prefix}label`][0];
+
+            if (-1 !== ['1', 'a_leg', 'inbound'].indexOf(sender.label)) {
+              opts.caller.aor = sender.aor ;
+              if (sender.name) opts.caller.name = sender.name;
+            }
+            else {
+              opts.callee.aor = sender.aor ;
+              if (sender.name) opts.callee.name = sender.name;
+            }
+          });
+
+          // if we dont have a participantstreamassoc then assume the first participant is the caller
+          if (!opts.caller.aor && !opts.callee.aor) {
+            let i = 0;
+            for (const part in participants) {
+              const p = participants[part];
+              if (0 === i && p.aor) {
+                opts.caller.aor = p.aor;
+                opts.caller.name = p.name;
+              }
+              else if (1 === i && p.aor) {
+                opts.callee.aor = p.aor;
+                opts.callee.name = p.name;
+              }
+              i++;
+            }
+          }
+
+          // now for Sonus (at least) we get the original from, to and call-id headers in a <callData/> element
+          // if so, this should take preference
+          const callData = parseCallData(prefix, obj);
+          if (callData) {
+            debug(`callData: ${JSON.stringify(callData)}`);
+            opts.originalCallId = callData.callid;
+
+            // caller
+            let r1 = /^(.*)(<sip.*)$/.exec(callData.fromhdr);
+            if (r1) {
+              const arr = /<(.*)>/.exec(r1[2]);
+              if (arr) {
+                const uri = parseUri(arr[1]);
+                const user = uri.user || 'anonymous';
+                opts.caller.aor = `sip:${user}@${uri.host}`;
+              }
+              const dname = r1[1].trim();
+              const arr2 = /"(.*)"/.exec(dname);
+              if (arr2) opts.caller.name = arr2[1];
+              else opts.caller.name = dname;
+            }
+            // callee
+            r1 = /^(.*)(<sip.*)$/.exec(callData.tohdr);
+            if (r1) {
+              const arr = /<(.*)>/.exec(r1[2]);
+              if (arr) {
+                const uri = parseUri(arr[1]);
+                opts.callee.aor = `sip:${uri.user}@${uri.host}`;
+              }
+              const dname = r1[1].trim();
+              const arr2 = /"(.*)"/.exec(dname);
+              if (arr2) opts.callee.name = arr2[1];
+              else opts.callee.name = dname;
+            }
+            debug(`opts.caller from callData: ${JSON.stringify(opts.caller)}`);
+            debug(`opts.callee from callData: ${JSON.stringify(opts.callee)}`);
+          }
+
+          if (opts.caller.aor && 0 !== opts.caller.aor.indexOf('sip:')) {
+            opts.caller.aor = 'sip:' + opts.caller.aor;
+          }
+          if (opts.callee.aor && 0 !== opts.callee.aor.indexOf('sip:')) {
+            opts.callee.aor = 'sip:' + opts.callee.aor;
+          }
+
+          if (opts.caller.aor) {
+            const uri = parseUri(opts.caller.aor);
+            opts.caller.number = uri.user;
+          }
+          if (opts.callee.aor) {
+            const uri = parseUri(opts.callee.aor);
+            opts.callee.number = uri.user;
+          }
+          opts.recordingSessionId = opts.recordingData[`${prefix}recording`][`${prefix}session`][0].$.session_id;
+        }
+      }
+      catch (err) {
+        reject(err);
+      }
+      debug(opts, 'payload parser results');
+      resolve(opts) ;
+    }) ;
+  }) ;
+};
+
+const createSipRecPayload = (sdp1, sdp2, logger) => {
+  const sdpObj = [];
+  sdpObj.push(transform.parse(sdp1));
+  sdpObj.push(transform.parse(sdp2));
+
+  //const arr1 = /^([^]+)(c=[^]+)t=[^]+(m=[^]+?)(a=[^]+)$/.exec(sdp1) ;
+  //const arr2 = /^([^]+)(c=[^]+)t=[^]+(m=[^]+?)(a=[^]+)$/.exec(sdp2) ;
+
+  debug(`sdp1:      ${sdp1}`);
+  debug(`objSdp[0]: ${JSON.stringify(sdpObj[0])}`);
+  debug(`sdp2:      ${sdp2}`);
+  debug(`objSdp[1]: ${JSON.stringify(sdpObj[1])}`);
+
+  if (!sdpObj[0] || !sdpObj[0].media.length) {
+    throw new Error(`Error parsing sdp1 into component parts: ${sdp1}`);
+  }
+  else if (!sdpObj[1] || !sdpObj[1].media.length)  {
+    throw new Error(`Error parsing sdp2 into component parts: ${sdp2}`);
+  }
+
+  if (!sdpObj[0].media[0].label) sdpObj[0].media[0].label = 1;
+  if (!sdpObj[1].media[0].label) sdpObj[1].media[0].label = 2;
+
+  //const aLabel = sdp1.includes('a=label:') ? '' : 'a=label:1\r\n';
+  //const bLabel = sdp2.includes('a=label:') ? '' : 'a=label:2\r\n';
+
+  sdpObj[0].media = sdpObj[0].media.concat(sdpObj[1].media);
+  const combinedSdp = transform.write(sdpObj[0])
+    .replace(/a=sendonly\r\n/g, '')
+    .replace(/a=direction:both\r\n/g, '');
+
+  debug(`combined ${combinedSdp}`);
+  /*
+  const combinedSdp = `${arr1[1]}t=0 0\r\n${arr1[2]}${arr1[3]}${arr1[4]}${aLabel}${arr2[3]}${arr2[4]}${bLabel}`
+    .replace(/a=sendonly\r\n/g, '')
+    .replace(/a=direction:both\r\n/g, '');
+  */
+  return combinedSdp;
+};
+
+module.exports = { parseSiprecPayload, createSipRecPayload } ;