Compare commits

..
6 Commits
Author SHA1 Message Date
Dave Horton a6582e71b5 fix ci test 2021-10-12 12:27:23 -04:00
Dave Horton 7b3d5c7a48 support for relaying dtmf via SIP INFO to FS 2021-10-12 09:42:06 -04:00
Dave Horton a0fcc3e335 update dep 2021-10-10 16:35:35 -04:00
Dave Horton b16519405a linting 2021-10-10 16:17:27 -04:00
Dave Horton b3ecb31d7f handle mute/unmute 2021-10-10 15:12:50 -04:00
Dave Horton ae968fcb50 feature: release media from freeswitch 2021-10-10 13:00:24 -04:00
25 changed files with 3011 additions and 5636 deletions
+1 -1
View File
@@ -11,7 +11,7 @@ jobs:
- uses: actions/checkout@v2
- uses: actions/setup-node@v1
with:
node-version: 14.x
node-version: 12
- run: npm ci
- run: npm run jslint
- run: npm test
-51
View File
@@ -1,51 +0,0 @@
name: Docker
on:
push:
# Publish `main` as Docker `latest` image.
branches:
- main
# Publish `v1.2.3` tags as releases.
tags:
- v*
env:
IMAGE_NAME: sbc-outbound
jobs:
push:
runs-on: ubuntu-latest
if: github.event_name == 'push'
steps:
- uses: actions/checkout@v2
- name: Build image
run: docker build . --file Dockerfile --tag $IMAGE_NAME
- name: Log into registry
run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
- name: Push image
run: |
IMAGE_ID=ghcr.io/${{ github.repository_owner }}/$IMAGE_NAME
# Change all uppercase to lowercase
IMAGE_ID=$(echo $IMAGE_ID | tr '[A-Z]' '[a-z]')
# Strip git ref prefix from version
VERSION=$(echo "${{ github.ref }}" | sed -e 's,.*/\(.*\),\1,')
# Strip "v" prefix from tag name
[[ "${{ github.ref }}" == "refs/tags/"* ]] && VERSION=$(echo $VERSION | sed -e 's/^v//')
# Use Docker `latest` tag convention
[ "$VERSION" == "main" ] && VERSION=latest
echo IMAGE_ID=$IMAGE_ID
echo VERSION=$VERSION
docker tag $IMAGE_NAME $IMAGE_ID:$VERSION
docker push $IMAGE_ID:$VERSION
-4
View File
@@ -1,4 +0,0 @@
#!/bin/sh
. "$(dirname "$0")/_/husky.sh"
npm run jslint
+9 -17
View File
@@ -1,23 +1,15 @@
FROM --platform=linux/amd64 node:18.12.1-alpine3.16 as base
RUN apk --update --no-cache add --virtual .builds-deps build-base python3
FROM node:alpine as builder
WORKDIR /opt/app/
COPY package.json ./
RUN npm install
RUN npm prune --production
FROM base as build
COPY package.json package-lock.json ./
RUN npm ci
COPY . .
FROM base
COPY --from=build /opt/app /opt/app/
FROM node:alpine as app
WORKDIR /opt/app
COPY . /opt/app
COPY --from=builder /opt/app/node_modules ./node_modules
ARG NODE_ENV
ENV NODE_ENV $NODE_ENV
CMD [ "node", "app.js" ]
CMD [ "npm", "start" ]
+28 -103
View File
@@ -6,20 +6,17 @@ assert.ok(process.env.JAMBONES_MYSQL_HOST &&
assert.ok(process.env.JAMBONES_REDIS_HOST, 'missing JAMBONES_REDIS_HOST env var');
assert.ok(process.env.DRACHTIO_PORT || process.env.DRACHTIO_HOST, 'missing DRACHTIO_PORT env var');
assert.ok(process.env.DRACHTIO_SECRET, 'missing DRACHTIO_SECRET env var');
assert.ok(process.env.JAMBONES_NETWORK_CIDR || process.env.K8S, 'missing JAMBONES_NETWORK_CIDR env var');
assert.ok(process.env.JAMBONES_NETWORK_CIDR, 'missing JAMBONES_NETWORK_CIDR env var');
const Srf = require('drachtio-srf');
const srf = new Srf('sbc-outbound');
const CIDRMatcher = require('cidr-matcher');
const {equalsIgnoreOrder, pingMsTeamsGateways, createHealthCheckApp, systemHealth} = require('./lib/utils');
const matcher = new CIDRMatcher([process.env.JAMBONES_NETWORK_CIDR]);
const opts = Object.assign({
timestamp: () => {return `, "time": "${new Date().toISOString()}"`;}
}, {level: process.env.JAMBONES_LOGLEVEL || 'info'});
const logger = require('pino')(opts);
const {
writeCallCount,
writeCallCountSP,
writeCallCountApp,
writeCdrs,
queryCdrs,
writeAlerts,
@@ -35,15 +32,11 @@ const CallSession = require('./lib/call-session');
const setNameRtp = `${(process.env.JAMBONES_CLUSTER_ID || 'default')}:active-rtp`;
const rtpServers = [];
const {
ping,
performLcr,
lookupAllTeamsFQDNs,
lookupAccountBySipRealm,
lookupAccountBySid,
lookupAccountCapacitiesBySid,
lookupSipGatewaysByCarrier,
lookupCarrierBySid,
queryCallLimits
lookupAccountCapacitiesBySid
} = require('@jambonz/db-helpers')({
host: process.env.JAMBONES_MYSQL_HOST,
user: process.env.JAMBONES_MYSQL_USER,
@@ -51,15 +44,7 @@ const {
database: process.env.JAMBONES_MYSQL_DATABASE,
connectionLimit: process.env.JAMBONES_MYSQL_CONNECTION_LIMIT || 10
}, logger);
const {
client: redisClient,
createHash,
retrieveHash,
incrKey,
decrKey,
retrieveSet,
isMemberOfSet
} = require('@jambonz/realtimedb-helpers')({
const {createHash, retrieveHash, incrKey, decrKey, retrieveSet} = require('@jambonz/realtimedb-helpers')({
host: process.env.JAMBONES_REDIS_HOST || 'localhost',
port: process.env.JAMBONES_REDIS_PORT || 6379
}, logger);
@@ -68,53 +53,36 @@ const activeCallIds = new Map();
srf.locals = {...srf.locals,
stats,
writeCallCount,
writeCallCountSP,
writeCallCountApp,
writeCdrs,
writeAlerts,
AlertType,
queryCdrs,
activeCallIds,
dbHelpers: {
ping,
performLcr,
lookupAllTeamsFQDNs,
lookupAccountBySipRealm,
lookupAccountBySid,
lookupAccountCapacitiesBySid,
lookupSipGatewaysByCarrier,
lookupCarrierBySid,
queryCallLimits
lookupAccountCapacitiesBySid
},
realtimeDbHelpers: {
createHash,
retrieveHash,
incrKey,
decrKey,
isMemberOfSet
decrKey
}
};
const {initLocals, checkLimits, route} = require('./lib/middleware')(srf, logger, {
host: process.env.JAMBONES_REDIS_HOST,
port: process.env.JAMBONES_REDIS_PORT || 6379
});
const ngProtocol = process.env.JAMBONES_NG_PROTOCOL || 'udp';
const ngPort = process.env.RTPENGINE_PORT || ('udp' === ngProtocol ? 22222 : 8080);
const {getRtpEngine, setRtpEngines} = require('@jambonz/rtpengine-utils')([], logger, {
//emitter: stats,
dtmfListenPort: process.env.DTMF_LISTEN_PORT || 22225,
protocol: ngProtocol
emitter: stats,
dtmfListenPort: process.env.DTMF_LISTEN_PORT || 22225
});
srf.locals.getRtpEngine = getRtpEngine;
if (process.env.DRACHTIO_HOST && !process.env.K8S) {
const cidrs = process.env.JAMBONES_NETWORK_CIDR
.split(',')
.map((s) => s.trim());
logger.info({cidrs}, 'internal network CIDRs');
const matcher = new CIDRMatcher(cidrs);
if (process.env.DRACHTIO_HOST) {
srf.connect({host: process.env.DRACHTIO_HOST, port: process.env.DRACHTIO_PORT, secret: process.env.DRACHTIO_SECRET });
srf.on('connect', (err, hp) => {
logger.info(`connected to drachtio listening on ${hp}`);
@@ -135,7 +103,6 @@ if (process.env.DRACHTIO_HOST && !process.env.K8S) {
});
}
else {
logger.info(`listening in outbound mode on port ${process.env.DRACHTIO_PORT}`);
srf.listen({port: process.env.DRACHTIO_PORT, secret: process.env.DRACHTIO_SECRET});
}
if (process.env.NODE_ENV === 'test') {
@@ -150,79 +117,35 @@ srf.invite((req, res) => {
session.connect();
});
if (process.env.K8S || process.env.HTTP_PORT) {
const PORT = process.env.HTTP_PORT || 3000;
const healthCheck = require('@jambonz/http-health-check');
/* update call stats periodically */
setInterval(() => {
stats.gauge('sbc.sip.calls.count', activeCallIds.size, ['direction:outbound']);
}, 5000);
const getCount = () => srf.locals.activeCallIds.size;
createHealthCheckApp(PORT, logger)
.then((app) => {
healthCheck({
app,
logger,
path: '/',
fn: getCount
});
healthCheck({
app,
logger,
path: '/system-health',
fn: systemHealth.bind(null, redisClient, ping, getCount)
});
return;
})
.catch((err) => {
logger.error({err}, 'Error creating health check server');
});
}
if ('test' !== process.env.NODE_ENV) {
/* update call stats periodically */
setInterval(() => {
stats.gauge('sbc.sip.calls.count', activeCallIds.size, ['direction:outbound']);
}, 5000);
}
const lookupRtpServiceEndpoints = (lookup, serviceName) => {
lookup(serviceName, {family: 4, all: true}, (err, addresses) => {
if (err) {
logger.error({err}, `Error looking up ${serviceName}`);
return;
}
logger.debug({addresses, rtpServers}, `dns lookup for ${serviceName} returned`);
const addrs = addresses.map((a) => a.address);
if (!equalsIgnoreOrder(addrs, rtpServers)) {
rtpServers.length = 0;
Array.prototype.push.apply(rtpServers, addrs);
logger.info({rtpServers}, 'rtpserver endpoints have been updated');
setRtpEngines(rtpServers.map((a) => `${a}:${ngPort}`));
}
});
const arrayCompare = (a, b) => {
if (a.length !== b.length) return false;
const uniqueValues = new Set([...a, ...b]);
for (const v of uniqueValues) {
const aCount = a.filter((e) => e === v).length;
const bCount = b.filter((e) => e === v).length;
if (aCount !== bCount) return false;
}
return true;
};
if (process.env.K8S_RTPENGINE_SERVICE_NAME) {
/* poll dns for endpoints every so often */
const arr = /^(.*):(\d+)$/.exec(process.env.K8S_RTPENGINE_SERVICE_NAME);
const svc = arr[1];
logger.info(`rtpengine(s) will be found at dns name: ${svc}`);
const {lookup} = require('dns');
lookupRtpServiceEndpoints(lookup, svc);
setInterval(lookupRtpServiceEndpoints.bind(null, lookup, svc), process.env.RTPENGINE_DNS_POLL_INTERVAL || 10000);
}
else if (process.env.JAMBONES_RTPENGINES) {
/* static list of rtpengines */
/* update rtpengines periodically */
if (process.env.JAMBONES_RTPENGINES) {
setRtpEngines([process.env.JAMBONES_RTPENGINES]);
}
else {
/* poll redis periodically for rtpengines that have registered via OPTIONS ping */
const getActiveRtpServers = async() => {
try {
const set = await retrieveSet(setNameRtp);
const newArray = Array.from(set);
logger.debug({newArray, rtpServers}, 'getActiveRtpServers');
if (!equalsIgnoreOrder(newArray, rtpServers)) {
if (!arrayCompare(newArray, rtpServers)) {
logger.info({newArray}, 'resetting active rtpengines');
setRtpEngines(newArray.map((a) => `${a}:${ngPort}`));
setRtpEngines(newArray.map((a) => `${a}:${process.env.RTPENGINE_PORT || 22222}`));
rtpServers.length = 0;
Array.prototype.push.apply(rtpServers, newArray);
}
@@ -230,12 +153,14 @@ else {
logger.error({err}, 'Error setting new rtpengines');
}
};
setInterval(() => {
getActiveRtpServers();
}, 30000);
getActiveRtpServers();
}
const {pingMsTeamsGateways} = require('./lib/utils');
pingMsTeamsGateways(logger, srf);
module.exports = {srf};
-29
View File
@@ -1,29 +0,0 @@
#!/usr/bin/env node
const bent = require('bent');
const getJSON = bent('json');
const PORT = process.env.HTTP_PORT || 3000;
const sleep = (ms) => {
return new Promise((resolve) => setTimeout(resolve, ms));
};
(async function() {
try {
do {
const obj = await getJSON(`http://127.0.0.1:${PORT}/`);
const {calls} = obj;
if (calls === 0) {
console.log('no calls on the system, we can exit');
process.exit(0);
}
else {
console.log(`waiting for ${calls} to exit..`);
}
await sleep(10000);
} while (1);
} catch (err) {
console.error(err, 'Error querying health endpoint');
process.exit(-1);
}
})();
+61 -349
View File
@@ -1,9 +1,8 @@
const Emitter = require('events');
const sdpTransform = require('sdp-transform');
const SrsClient = require('@jambonz/siprec-client-utils');
const {makeRtpEngineOpts, nudgeCallCounts} = require('./utils');
const {makeRtpEngineOpts, makeCallCountKey} = require('./utils');
const {forwardInDialogRequests} = require('drachtio-fn-b2b-sugar');
const {SipError, stringifyUri, parseUri} = require('drachtio-srf');
const {SipError} = require('drachtio-srf');
const {parseUri} = require('drachtio-srf');
const debug = require('debug')('jambonz:sbc-outbound');
const makeInviteInProgressKey = (callid) => `sbc-out-iip${callid}`;
@@ -13,17 +12,10 @@ const makeInviteInProgressKey = (callid) => `sbc-out-iip${callid}`;
*/
const createBLegFromHeader = (req, teams) => {
const from = req.getParsedHeader('From');
const host = teams ? req.get('X-MS-Teams-Tenant-FQDN') : 'localhost';
const uri = parseUri(from.uri);
let user = uri.user || 'anonymous';
let host = 'localhost';
if (teams) {
host = req.get('X-MS-Teams-Tenant-FQDN');
}
else if (req.has('X-Preferred-From-User') || req.has('X-Preferred-From-Host')) {
user = req.get('X-Preferred-From-User') || user;
host = req.get('X-Preferred-From-Host') || host;
}
return `sip:${user}@${host}`;
if (uri && uri.user) return `sip:${uri.user}@${host}`;
return `sip:anonymous@${host}`;
};
const createBLegToHeader = (req, teams) => {
const to = req.getParsedHeader('To');
@@ -40,13 +32,11 @@ const initCdr = (srf, req) => {
const to = arr ? arr[1] : req.calledNumber;
arr = regex.exec(req.callingNumber);
const from = arr ? arr[1] : req.callingNumber;
const applicationSid = req.get('X-Application-Sid');
return {
account_sid: req.get('X-Account-Sid'),
call_sid: req.get('X-Call-Sid'),
sip_callid: req.get('Call-ID'),
...(applicationSid && {application_sid: applicationSid}),
from,
to,
duration: 0,
@@ -54,58 +44,36 @@ const initCdr = (srf, req) => {
attempted_at: Date.now(),
direction: 'outbound',
host: srf.locals.sipAddress,
remote_host: uri.host,
trace_id: req.get('X-Trace-ID') || '00000000000000000000000000000000'
remote_host: uri.host
};
};
const updateRtpEngineFlags = (sdp, opts) => {
try {
const parsed = sdpTransform.parse(sdp);
const codec = parsed.media[0].rtp[0].codec;
if (['PCMU', 'PCMA'].includes(codec)) opts.flags.push(`codec-accept-${codec}`);
} catch (err) {}
return opts;
};
class CallSession extends Emitter {
constructor(logger, req, res) {
super();
this.req = req;
this.res = res;
this.srf = req.srf;
this.performLcr = this.srf.locals.dbHelpers.performLcr;
this.logger = logger.child({callId: req.get('Call-ID')});
this.useWss = req.locals.registration && req.locals.registration.protocol === 'wss';
this.stats = this.srf.locals.stats;
this.activeCallIds = this.srf.locals.activeCallIds;
this.writeCdrs = this.srf.locals.writeCdrs;
this.incrKey = req.srf.locals.realtimeDbHelpers.incrKey;
this.decrKey = req.srf.locals.realtimeDbHelpers.decrKey;
const {performLcr, lookupCarrierBySid, lookupSipGatewaysByCarrier} = this.srf.locals.dbHelpers;
this.performLcr = performLcr;
this.lookupCarrierBySid = lookupCarrierBySid;
this.lookupSipGatewaysByCarrier = lookupSipGatewaysByCarrier;
this._mediaReleased = false;
this.callCountKey = makeCallCountKey(req.locals.account_sid);
}
get account_sid() {
return this.req.locals.account_sid;
}
get application_sid() {
return this.req.locals.application_sid;
}
get privateSipAddress() {
return this.srf.locals.privateSipAddress;
}
get isMediaReleased() {
return this._mediaReleased;
}
async connect() {
const teams = this.teams = this.req.locals.target === 'teams';
const engine = this.srf.locals.getRtpEngine();
@@ -122,12 +90,8 @@ class CallSession extends Emitter {
unblockMedia,
blockDTMF,
unblockDTMF,
playDTMF,
subscribeDTMF,
unsubscribeDTMF,
subscribeRequest,
subscribeAnswer,
unsubscribe
unsubscribeDTMF
} = engine;
const {createHash, retrieveHash} = this.srf.locals.realtimeDbHelpers;
this.offer = offer;
@@ -137,12 +101,8 @@ class CallSession extends Emitter {
this.unblockMedia = unblockMedia;
this.blockDTMF = blockDTMF;
this.unblockDTMF = unblockDTMF;
this.playDTMF = playDTMF;
this.subscribeDTMF = subscribeDTMF;
this.unsubscribeDTMF = unsubscribeDTMF;
this.subscribeRequest = subscribeRequest;
this.subscribeAnswer = subscribeAnswer;
this.unsubscribe = unsubscribe;
this.rtpEngineOpts = makeRtpEngineOpts(this.req, false, this.useWss || teams, teams);
this.rtpEngineResource = {destroy: this.del.bind(null, this.rtpEngineOpts.common)};
@@ -161,19 +121,14 @@ class CallSession extends Emitter {
};
if (this.req.locals.registration) {
debug(`sending call to registered user ${JSON.stringify(this.req.locals.registration)}`);
debug(`sending call to user ${JSON.stringify(this.req.locals.registration)}`);
const contact = this.req.locals.registration.contact;
let destUri = contact;
if (this.req.has('X-Override-To')) {
const dest = this.req.get('X-Override-To');
const uri = parseUri(contact);
uri.user = dest;
destUri = stringifyUri(uri);
this.logger.info(`overriding destination user with ${dest}, so final uri is ${destUri}`);
if (contact.includes('transport=ws')) {
uris = [contact];
}
uris = [destUri];
if (!contact.includes('transport=ws')) {
else {
proxy = this.req.locals.registration.proxy;
uris = [this.req.uri];
}
}
else if (this.req.locals.target === 'forward') {
@@ -194,76 +149,39 @@ class CallSession extends Emitter {
else {
debug('calling lcr');
try {
/* was a specific carrier requested */
const voip_carrier_sid = this.req.get('X-Requested-Carrier-Sid');
if (voip_carrier_sid) {
const vc = await this.lookupCarrierBySid(voip_carrier_sid);
const gateways = await this.lookupSipGatewaysByCarrier(voip_carrier_sid);
const gws = (gateways || [])
.filter((gw) => gw.outbound);
if (gws.length) {
uris = [];
gws.forEach((o) => {
const prefix = vc.tech_prefix;
const hostport = !o.port || 5060 === o.port ? o.ipv4 : `${o.ipv4}:${o.port}`;
const prependPlus = vc.e164_leading_plus && !this.req.calledNumber.startsWith('0');
const u = `sip:${prefix ? prefix : ''}${prependPlus ? '+' : ''}${this.req.calledNumber}@${hostport}`;
const obj = {
name: vc.name,
diversion: vc.diversion,
hostport
};
if (vc.register_username && vc.register_password) {
obj.auth = {
username: vc.register_username,
password: vc.register_password
};
}
mapGateways.set(u, obj);
uris.push(u);
});
this.logger.debug({uris, voip_carrier_sid}, 'selected outbound gateways for requested carrier');
}
else {
this.logger.info({voip_carrier_sid}, 'no outbound gateways found for requested carrier');
}
}
if (mapGateways.size === 0) {
/**
* We normalize the called number by removing a leading + before sending it to LCR..
* but LCR will return us an array of sip uris, with leading + for carriers that require it
*/
const routableNumber = this.req.calledNumber.startsWith('+') ?
this.req.calledNumber.slice(1) :
this.req.calledNumber;
const gateways = await this.performLcr(routableNumber, this.account_sid);
if (!gateways || gateways.length === 0) throw new Error('no routes found');
debug(`got gateways: ${JSON.stringify(gateways)}`);
gateways.forEach((gw) => mapGateways.set(gw.uri, {
name: gw.name,
auth: gw.auth,
diversion: gw.diversion,
hostport: gw.hostport
}));
uris = gateways.map((gw) => gw.uri);
}
/**
* We normalize the called number by removing a leading + before sending it to LCR..
* but LCR will return us an array of sip uris, with leading + for carriers that require it
*/
const routableNumber = this.req.calledNumber.startsWith('+') ?
this.req.calledNumber.slice(1) :
this.req.calledNumber;
const gateways = await this.performLcr(routableNumber, this.account_sid);
if (!gateways || gateways.length === 0) throw new Error('no routes found');
debug(`got gateways: ${JSON.stringify(gateways)}`);
gateways.forEach((gw) => mapGateways.set(gw.uri, {
name: gw.name,
auth: gw.auth,
diversion: gw.diversion,
hostport: gw.hostport
}));
uris = gateways.map((gw) => gw.uri);
} catch (err) {
debug(err);
this.logger.error(err, 'Error performing lcr');
this.res.send(488);
return this.srf.endSession(this.req);
return this.res.send(488);
}
debug(`sending call to PSTN ${uris}`);
}
// rtpengine 'offer'
const opts = updateRtpEngineFlags(this.req.body, {
const opts = {
...this.rtpEngineOpts.common,
...this.rtpEngineOpts.uac.mediaOpts,
'from-tag': this.rtpEngineOpts.uas.tag,
direction: ['private', 'public'],
sdp: this.req.body
});
};
const response = await this.offer(opts);
debug(`response from rtpengine to offer ${JSON.stringify(response)}`);
this.logger.debug({offer: opts, response}, 'initial offer to rtpengine');
@@ -319,25 +237,9 @@ class CallSession extends Emitter {
const {uas, uac} = await this.srf.createB2BUA(this.req, this.res, uri, {
proxy,
passFailure,
proxyRequestHeaders: [
'all',
'-X-MS-Teams-FQDN',
'-X-MS-Teams-Tenant-FQDN',
'-X-Trace-ID',
'-Allow',
'-Session-Expires',
'-X-Requested-Carrier-Sid',
'-X-Jambonz-Routing',
'-X-Jambonz-FS-UUID',
'-X-Preferred-From-User',
'X-Preferred-From-Host',
'-X-Jambonz-FS-UUID',
],
proxyResponseHeaders: [
'all',
'-Allow',
'-Session-Expires'
],
proxyRequestHeaders: ['all', '-X-MS-Teams-FQDN', '-X-MS-Teams-Tenant-FQDN', 'X-CID', '-Allow',
'-Session-Expires', 'Min-SE'],
proxyResponseHeaders: ['all', '-Allow', '-Session-Expires'],
headers: hdrs,
responseHeaders,
auth: gw ? gw.auth : undefined,
@@ -363,7 +265,6 @@ class CallSession extends Emitter {
}
}, {
cbRequest: async(err, inv) => {
if (err) return this.logger.info({err}, 'CallSession error sending INVITE');
let trunk = gw ? gw.name : null;
if (!trunk) {
if (teams) trunk = 'Microsoft Teams';
@@ -373,9 +274,7 @@ class CallSession extends Emitter {
if (!this.req.locals.account.disable_cdrs) {
this.req.locals.cdr = {
...initCdr(this.req.srf, inv),
service_provider_sid: this.req.locals.service_provider_sid,
account_sid: this.req.locals.account_sid,
...(this.req.locals.application_sid && {application_sid: this.req.locals.application_sid}),
trunk
};
}
@@ -397,7 +296,6 @@ class CallSession extends Emitter {
// successfully connected
this.logger.info(`call connected to ${uri}`);
this.connectedUri = uri;
debug('call connected');
this.emit('connected', uri);
@@ -411,16 +309,14 @@ class CallSession extends Emitter {
!(err instanceof SipError) || // unexpected error
err.status === 487) { // caller hung up
const abandoned = err.message && err.message.includes('rtpengine failed: Unknown call-id');
const abandoned = err.message.includes('rtpengine failed: Unknown call-id');
const status = err.status || (abandoned ? 487 : 500);
if (err instanceof SipError) this.logger.info(`final call failure ${status}`);
else if (!abandoned) this.logger.error(err, 'unexpected call failure');
debug(`got final outdial error: ${err}`);
if (!passFailure) this.res.send(status);
this.emit('failed');
this.rtpEngineResource.destroy()
.catch((err) => this.logger.info({err}, 'Error destroying rtpe after failure'));
this.srf.endSession(this.req);
this.rtpEngineResource.destroy();
const tags = ['accepted:no', `sipStatus:${status}`];
this.stats.increment('sbc.originations', tags);
@@ -428,10 +324,9 @@ class CallSession extends Emitter {
this.writeCdrs({...this.req.locals.cdr,
terminated_at: Date.now(),
termination_reason: 487 === status ? 'caller abandoned' : 'failed',
sip_status: status
sip_status: status,
}).catch((err) => this.logger.error({err}, 'Error writing cdr for call failure'));
}
return;
}
else {
this.logger.info(`got ${err.status}, cranking back to next destination`);
@@ -439,9 +334,8 @@ class CallSession extends Emitter {
}
}
} catch (err) {
if ('abandonded' !== err.message) this.logger.error(err, `Error setting up outbound call to: ${uris}`);
if ('abandonded' !== err.message) this.logger.error(err, `Error setting up outbonund call to: ${uris}`);
this.emit('failed');
this.srf.endSession(this.req);
this.rtpEngineResource.destroy();
}
}
@@ -461,28 +355,20 @@ class CallSession extends Emitter {
this.uas = uas;
this.uac = uac;
[uas, uac].forEach((dlg) => {
dlg.on('destroy', async() => {
const other = dlg.other;
dlg.on('destroy', () => {
this.logger.info('call ended');
this.rtpEngineResource.destroy();
this.activeCallIds.delete(this.req.get('Call-ID'));
this.unsubscribeDTMF(this.logger, this.req.get('Call-ID'), this.rtpEngineOpts.uac.tag);
try {
await other.destroy();
} catch (err) {}
dlg.other.destroy();
const trackingOn = process.env.JAMBONES_TRACK_ACCOUNT_CALLS ||
process.env.JAMBONES_TRACK_SP_CALLS ||
process.env.JAMBONES_TRACK_APP_CALLS;
if (process.env.JAMBONES_HOSTING || trackingOn) {
const {writeCallCount, writeCallCountSP, writeCallCountApp} = this.req.srf.locals;
await nudgeCallCounts(this.logger, {
service_provider_sid: this.service_provider_sid,
account_sid: this.account_sid,
application_sid: this.application_sid
}, this.decrKey, {writeCallCountSP, writeCallCount, writeCallCountApp})
.catch((err) => this.logger.error(err, 'Error decrementing call counts'));
}
this.decrKey(this.callCountKey)
.then((count) => {
this.logger.debug(`after hangup there are ${count} active calls for this account`);
debug(`after hangup there are ${count} active calls for this account`);
return;
})
.catch((err) => this.logger.error({err}, 'Error decrementing call count'));
/* write cdr for connected call */
if (this.req.locals.cdr) {
@@ -495,15 +381,6 @@ class CallSession extends Emitter {
duration: Math.floor((now - callStart) / 1000)
}).catch((err) => this.logger.error({err}, 'Error writing cdr for completed call'));
}
/* de-link the 2 Dialogs for GC */
dlg.removeAllListeners();
other.removeAllListeners();
dlg.other = null;
other.other = null;
this.logger.info(`call ended with normal termination, there are ${this.activeCallIds.size} active`);
this.srf.endSession(this.req);
});
});
@@ -514,7 +391,6 @@ class CallSession extends Emitter {
uac.on('modify', this._onReinvite.bind(this, uac));
uas.on('refer', this._onFeatureServerTransfer.bind(this, uas));
uac.on('refer', this._onRefer.bind(this, uac));
uas.on('info', this._onInfo.bind(this, uas));
uac.on('info', this._onInfo.bind(this, uac));
@@ -523,23 +399,6 @@ class CallSession extends Emitter {
forwardInDialogRequests(uac, ['notify', 'options', 'message']);
}
async _onRefer(dlg, req, res) {
/* REFER coming in from a sip device, forward to feature server */
try {
const response = await dlg.other.request({
method: 'REFER',
headers: {
'Refer-To': req.get('Refer-To'),
'Referred-By': req.get('Referred-By'),
'User-Agent': req.get('User-Agent')
}
});
res.send(response.status, response.reason);
} catch (err) {
this.logger.error({err}, 'CallSession:_onRefer: error handling incoming REFER');
}
}
async _onDTMF(dlg, payload) {
this.logger.info({payload}, '_onDTMF');
try {
@@ -571,17 +430,11 @@ Duration=${payload.duration} `
async _onReinvite(dlg, req, res) {
try {
const reason = req.get('X-Reason');
const isReleasingMedia = reason && dlg.type === 'uas' && ['release-media', 'anchor-media'].includes(reason);
const fromTag = dlg.type === 'uas' ? this.rtpEngineOpts.uas.tag : this.rtpEngineOpts.uac.tag;
const toTag = dlg.type === 'uas' ? this.rtpEngineOpts.uac.tag : this.rtpEngineOpts.uas.tag;
const offerMedia = dlg.type === 'uas' ? this.rtpEngineOpts.uac.mediaOpts : this.rtpEngineOpts.uas.mediaOpts;
const answerMedia = dlg.type === 'uas' ? this.rtpEngineOpts.uas.mediaOpts : this.rtpEngineOpts.uac.mediaOpts;
const direction = dlg.type === 'uas' ? ['private', 'public'] : ['public', 'private'];
if (isReleasingMedia) {
if (!offerMedia.flags.includes('port latching')) offerMedia.flags.push('port latching');
if (!offerMedia.flags.includes('asymmetric')) offerMedia.flags.push('asymmetric');
offerMedia.flags = offerMedia.flags.filter((f) => f !== 'media handover');
}
let opts = {
...this.rtpEngineOpts.common,
...offerMedia,
@@ -590,29 +443,22 @@ Duration=${payload.duration} `
direction,
sdp: req.body,
};
// DH: this was restarting ICE, which we don't want to do
//if (reason && opts.flags && !opts.flags.includes('reset')) opts.flags.push('reset');
if (reason) opts.flags.push('reset');
let response = await this.offer(opts);
if ('ok' !== response.result) {
res.send(488);
throw new Error(`_onReinvite: rtpengine failed: offer: ${JSON.stringify(response)}`);
}
this.logger.debug({opts, response}, 'CallSession:_onReinvite: (offer)');
/* if this is a re-invite from the FS to change media anchoring, avoid sending the reinvite out */
let sdp;
if (isReleasingMedia) {
if (reason && dlg.type === 'uas' && ['release-media', 'anchor-media'].includes(reason)) {
this.logger.info(`got a reinvite from FS to ${reason}`);
sdp = dlg.other.remote.sdp;
if (!answerMedia.flags.includes('port latching')) answerMedia.flags.push('port latching');
if (!answerMedia.flags.includes('asymmetric')) answerMedia.flags.push('asymmetric');
answerMedia.flags = answerMedia.flags.filter((f) => f !== 'media handover');
this._mediaReleased = 'release-media' === reason;
}
else {
sdp = await dlg.other.modify(response.sdp);
this.logger.info({sdp}, 'CallSession:_onReinvite: got sdp from 200 OK to invite we sent');
}
opts = {
...this.rtpEngineOpts.common,
@@ -626,7 +472,6 @@ Duration=${payload.duration} `
res.send(488);
throw new Error(`_onReinvite: rtpengine failed: ${JSON.stringify(response)}`);
}
this.logger.debug({opts, sdp: response.sdp}, 'CallSession:_onReinvite: (answer) sending back upstream');
res.send(200, {body: response.sdp});
} catch (err) {
this.logger.error(err, 'Error handling reinvite');
@@ -635,7 +480,6 @@ Duration=${payload.duration} `
async _onInfo(dlg, req, res) {
try {
const contentType = req.get('Content-Type');
if (dlg.type === 'uas' && req.has('X-Reason')) {
const toTag = this.rtpEngineOpts.uac.tag;
const reason = req.get('X-Reason');
@@ -654,123 +498,6 @@ Duration=${payload.duration} `
const response = Promise.all([this.unblockMedia(opts), this.unblockDTMF(opts)]);
this.logger.info({response}, `_onInfo: response to rtpengine command for ${reason}`);
}
else if (reason.includes('CallRecording')) {
let succeeded = false;
if (reason === 'startCallRecording') {
const from = this.req.getParsedHeader('From');
const to = this.req.getParsedHeader('To');
const aorFrom = from.uri;
const aorTo = to.uri;
this.logger.info({to, from}, 'startCallRecording request for an outbound call');
const srsUrl = req.get('X-Srs-Url');
const srsRecordingId = req.get('X-Srs-Recording-ID');
const callSid = req.get('X-Call-Sid');
const accountSid = req.get('X-Account-Sid');
const applicationSid = req.get('X-Application-Sid');
if (this.srsClient) {
res.send(400);
this.logger.info('discarding duplicate startCallRecording request for a call');
return;
}
if (!srsUrl) {
this.logger.info('startCallRecording request is missing X-Srs-Url header');
res.send(400);
return;
}
this.srsClient = new SrsClient(this.logger, {
srf: dlg.srf,
direction: 'outbound',
originalInvite: this.req,
callingNumber: this.req.callingNumber,
calledNumber: this.req.calledNumber,
srsUrl,
srsRecordingId,
callSid,
accountSid,
applicationSid,
rtpEngineOpts: this.rtpEngineOpts,
toTag,
aorFrom,
aorTo,
subscribeRequest: this.subscribeRequest,
subscribeAnswer: this.subscribeAnswer,
del: this.del,
blockMedia: this.blockMedia,
unblockMedia: this.unblockMedia,
unsubscribe: this.unsubscribe
});
try {
succeeded = await this.srsClient.start();
} catch (err) {
this.logger.error({err}, 'Error starting SipRec call recording');
}
}
else if (reason === 'stopCallRecording') {
if (!this.srsClient) {
res.send(400);
this.logger.info('discarding stopCallRecording request because we are not recording');
return;
}
try {
succeeded = await this.srsClient.stop();
} catch (err) {
this.logger.error({err}, 'Error stopping SipRec call recording');
}
this.srsClient = null;
}
else if (reason === 'pauseCallRecording') {
if (!this.srsClient || this.srsClient.paused) {
this.logger.info('discarding invalid pauseCallRecording request');
res.send(400);
return;
}
succeeded = await this.srsClient.pause();
}
else if (reason === 'resumeCallRecording') {
if (!this.srsClient || !this.srsClient.paused) {
res.send(400);
this.logger.info('discarding invalid resumeCallRecording request');
return;
}
succeeded = await this.srsClient.resume();
}
res.send(succeeded ? 200 : 503);
}
}
else if (dlg.type === 'uac' && ['application/dtmf-relay', 'application/dtmf'].includes(contentType)) {
const arr = /Signal=\s*([1-9#*])/.exec(req.body);
if (!arr) {
this.logger.info({body: req.body}, '_onInfo: invalid INFO dtmf request');
throw new Error(`_onInfo: no dtmf in body for ${contentType}`);
}
const code = arr[1];
const arr2 = /Duration=\s*(\d+)/.exec(req.body);
const duration = arr2 ? arr2[1] : 250;
if (this.isMediaReleased) {
/* just relay on to the feature server */
this.logger.info({code, duration}, 'got SIP INFO DTMF from caller, relaying to feature server');
this._onDTMF(dlg.other, {event: code, duration})
.catch((err) => this.logger.info({err}, 'Error relaying DTMF to feature server'));
res.send(200);
}
else {
/* else convert SIP INFO to RFC 2833 telephony events */
this.logger.info({code, duration}, 'got SIP INFO DTMF from caller, converting to RFC 2833');
const opts = {
...this.rtpEngineOpts.common,
'from-tag': this.rtpEngineOpts.uac.tag,
code,
duration
};
const response = await this.playDTMF(opts);
if ('ok' !== response.result) {
this.logger.info({response}, `rtpengine playDTMF failed with ${JSON.stringify(response)}`);
throw new Error('rtpengine failed: answer');
}
res.send(200);
}
}
else {
const response = await dlg.other.request({
@@ -795,30 +522,16 @@ Duration=${payload.duration} `
this.logger.info({uri, referTo}, 'received REFER from feature server');
const arr = /context-(.*)/.exec(uri.user);
if (!arr) {
/* call transfer requested */
const referredBy = req.getParsedHeader('Referred-By');
if (!referredBy) return res.send(400);
const u = parseUri(referredBy.uri);
const farEnd = parseUri(this.connectedUri);
uri.host = farEnd.host;
uri.port = farEnd.port;
const response = await this.uac.request({
method: 'REFER',
headers: {
'Refer-To': stringifyUri(uri),
'Referred-By': stringifyUri(u)
}
});
return res.send(response.status);
this.logger.info(`invalid Refer-To header: ${referTo.uri}`);
return res.send(501);
}
res.send(202);
// invite to new fs
const headers = {
...(req.has('X-Retain-Call-Sid') && {'X-Retain-Call-Sid': req.get('X-Retain-Call-Sid')}),
...(req.has('X-Account-Sid') && {'X-Account-Sid': req.get('X-Account-Sid')})
};
const headers = {};
if (req.has('X-Retain-Call-Sid')) {
Object.assign(headers, {'X-Retain-Call-Sid': req.get('X-Retain-Call-Sid')});
}
const dlg = await this.srf.createUAC(referTo.uri, {localSdp: dlg.local.sdp, headers});
this.uas = dlg;
this.uas.other = this.uac;
@@ -830,7 +543,6 @@ Duration=${payload.duration} `
this.rtpEngineResource.destroy();
this.activeCallIds.delete(this.req.get('Call-ID'));
this.uas.other.destroy();
this.srf.endSession(this.req);
});
// modify rtpengine to stream to new feature server
+69 -132
View File
@@ -1,68 +1,36 @@
const debug = require('debug')('jambonz:sbc-outbound');
const parseUri = require('drachtio-srf').parseUri;
const Registrar = require('@jambonz/mw-registrar');
const {selectHostPort, nudgeCallCounts} = require('./utils');
const FS_UUID_SET_NAME = 'fsUUIDs';
const {selectHostPort, makeCallCountKey} = require('./utils');
const isLocalUri = (host, req) => {
debug({hostport: req.server.hostport}, `is ${host} local?`);
return req.server.hostport.includes(host);
};
module.exports = (srf, logger, opts) => {
const {incrKey, decrKey, isMemberOfSet} = srf.locals.realtimeDbHelpers;
const {incrKey, decrKey} = srf.locals.realtimeDbHelpers;
const {stats} = srf.locals;
const registrar = new Registrar(opts);
const {
lookupAccountCapacitiesBySid,
lookupAccountBySid,
queryCallLimits
lookupAccountBySid
} = srf.locals.dbHelpers;
const initLocals = async(req, res, next) => {
req.locals = req.locals || {};
const callId = req.get('Call-ID');
req.locals.account_sid = req.get('X-Account-Sid');
req.locals.application_sid = req.get('X-Application-Sid');
const traceId = req.locals.trace_id = req.get('X-Trace-ID');
req.locals.logger = logger.child({
callId,
traceId,
account_sid:
req.locals.account_sid});
req.locals.logger = logger.child({callId, account_sid: req.locals.account_sid});
if (!req.locals.account_sid) {
logger.info('missing X-Account-Sid on outbound call');
res.send(403, {
return res.send(403, {
headers: {
'X-Reason': 'missing X-Account-Sid'
}
});
return req.srf.endSession(req);
}
/* must come from a valid FS */
if (!req.has('X-Jambonz-Routing')) {
logger.info({msg: req.msg}, 'missing X-Jambonz-Routing header');
res.send(403, {
headers: {
'X-Reason': 'missing required jambonz headers'
}
});
return req.srf.endSession(req);
}
if (process.env.K8S) {
/* for K8S we do not use JAMBONES_CIDR so we must validate the sender by uuid FS creates */
const fsUUID = req.get('X-Jambonz-FS-UUID');
try {
const exists = await isMemberOfSet(FS_UUID_SET_NAME, fsUUID);
if (!exists || !fsUUID) {
res.send(403, {
headers: {
'X-Reason': `missing or invalid FS-UUID ${fsUUID}`
}
});
return req.srf.endSession(req);
}
} catch (err) {
res.send(500);
return req.srf.endSession(req);
}
}
stats.increment('sbc.invites', ['direction:outbound']);
@@ -77,37 +45,34 @@ module.exports = (srf, logger, opts) => {
try {
req.locals.account = await lookupAccountBySid(req.locals.account_sid);
req.locals.service_provider_sid = req.locals.account.service_provider_sid;
} catch (err) {
req.locals.logger.error({err}, `Error looking up account sid ${req.locals.account_sid}`);
res.send(500);
return req.srf.endSession(req);
return res.send(500);
}
next();
};
const checkLimits = async(req, res, next) => {
const {logger, account_sid, service_provider_sid, application_sid} = req.locals;
const trackingOn = process.env.JAMBONES_TRACK_ACCOUNT_CALLS ||
process.env.JAMBONES_TRACK_SP_CALLS ||
process.env.JAMBONES_TRACK_APP_CALLS;
if (!process.env.JAMBONES_HOSTING && !trackingOn) {
logger.debug('tracking is off, skipping call limit checks');
return next(); // skip
}
const {writeCallCount, writeCallCountSP, writeCallCountApp, writeAlerts, AlertType} = req.srf.locals;
const {logger, account_sid} = req.locals;
const {writeAlerts, AlertType} = req.srf.locals;
const key = makeCallCountKey(account_sid);
try {
/* increment the call count */
const calls = await incrKey(key);
debug(`checkLimits: call count is now ${calls}`);
/* decrement count if INVITE is later rejected */
res.once('end', async({status}) => {
res.once('end', ({status}) => {
if (status > 200) {
nudgeCallCounts(logger, {
service_provider_sid,
account_sid,
application_sid
}, decrKey, {writeCallCountSP, writeCallCount, writeCallCountApp})
.catch((err) => logger.error(err, 'Error decrementing call counts'));
debug('checkLimits: decrementing call count due to rejection');
decrKey(key)
.then((count) => {
logger.debug({key}, `after rejection there are ${count} active calls for this account`);
debug({key}, `after rejection there are ${count} active calls for this account`);
return;
})
.catch((err) => logger.error({err}, 'checkLimits: decrKey err'));
const tags = ['accepted:no', `sipStatus:${status}`];
stats.increment('sbc.originations', tags);
}
@@ -117,13 +82,6 @@ module.exports = (srf, logger, opts) => {
}
});
/* increment the call count */
const {callsSP, calls} = await nudgeCallCounts(logger, {
service_provider_sid,
account_sid,
application_sid
}, incrKey, {writeCallCountSP, writeCallCount, writeCallCountApp});
/* compare to account's limit, though avoid db hit when call count is low */
const minLimit = process.env.MIN_CALL_LIMIT ?
parseInt(process.env.MIN_CALL_LIMIT) :
@@ -132,54 +90,22 @@ module.exports = (srf, logger, opts) => {
const capacities = await lookupAccountCapacitiesBySid(account_sid);
const limit = capacities.find((c) => c.category == 'voice_call_session');
if (limit) {
const limit_sessions = limit.quantity;
if (calls > limit_sessions) {
logger.info({calls, limit_sessions}, 'checkLimits: limits exceeded');
writeAlerts({
alert_type: AlertType.ACCOUNT_CALL_LIMIT,
service_provider_sid,
account_sid,
count: limit_sessions
}).catch((err) => logger.info({err}, 'checkLimits: error writing alert'));
res.send(503, 'Maximum Calls In Progress');
return req.srf.endSession(req);
}
if (!limit) {
logger.debug('checkLimits: no call limits specified');
return next();
}
else if (trackingOn) {
const {account_limit, sp_limit} = await queryCallLimits(service_provider_sid, account_sid);
if (process.env.JAMBONES_TRACK_ACCOUNT_CALLS && account_limit > 0 && calls > account_limit) {
logger.info({calls, account_limit}, 'checkLimits: account limits exceeded');
writeAlerts({
alert_type: AlertType.ACCOUNT_CALL_LIMIT,
service_provider_sid: service_provider_sid,
account_sid,
count: calls
}).catch((err) => logger.info({err}, 'checkLimits: error writing alert'));
res.send(503, 'Max Account Calls In Progress', {
headers: {
'X-Account-Sid': account_sid,
'X-Call-Limit': account_limit
}
});
return req.srf.endSession(req);
}
if (process.env.JAMBONES_TRACK_SP_CALLS && sp_limit > 0 && callsSP > sp_limit) {
logger.info({callsSP, sp_limit}, 'checkLimits: service provider limits exceeded');
writeAlerts({
alert_type: AlertType.SP_CALL_LIMIT,
service_provider_sid: service_provider_sid,
count: callsSP
}).catch((err) => logger.info({err}, 'checkLimits: error writing alert'));
res.send(503, 'Max Service Provider Calls In Progress', {
headers: {
'X-Service-Provider-Sid': service_provider_sid,
'X-Call-Limit': sp_limit
}
});
return req.srf.endSession(req);
}
const limit_sessions = limit.quantity;
if (calls > limit_sessions) {
debug(`checkLimits: limits exceeded: call count ${calls}, limit ${limit_sessions}`);
logger.info({calls, limit_sessions}, 'checkLimits: limits exceeded');
writeAlerts({
alert_type: AlertType.CALL_LIMIT,
account_sid,
count: limit_sessions
}).catch((err) => logger.info({err}, 'checkLimits: error writing alert'));
res.send(503, 'Maximum Calls In Progress');
}
next();
} catch (err) {
@@ -193,7 +119,6 @@ module.exports = (srf, logger, opts) => {
const {lookupAccountBySipRealm} = req.srf.locals.dbHelpers;
logger.info(`received outbound INVITE to ${req.uri} from server at ${req.server.hostport}`);
const uri = parseUri(req.uri);
const desiredRouting = req.get('X-Jambonz-Routing');
if (!uri || !uri.user || !uri.host) {
logger.info({uri: req.uri}, 'invalid request-uri on outbound call, rejecting');
@@ -202,17 +127,23 @@ module.exports = (srf, logger, opts) => {
'X-Reason': 'invalid request-uri'
}
});
return req.srf.endSession(req);
return;
}
const aor = `${uri.user}@${uri.host}`;
let reg;
const dotDecimalHost = /^[0-9\.]+$/.test(uri.host);
debug(`received outbound INVITE to ${req.calledNumber} from server at ${req.server.hostport}`);
if ('teams' === desiredRouting) {
if (req.has('X-MS-Teams-FQDN') && req.has('X-MS-Teams-Tenant-FQDN')) {
logger.debug('This is a call to ms teams');
req.locals.target = 'teams';
return next();
}
else if ('user' === desiredRouting) {
const aor = `${uri.user}@${uri.host}`;
const reg = await registrar.query(aor);
else if (!dotDecimalHost) {
// uri host is not a dot-decimal address, so try to look up user
logger.debug(`searching for registered user ${aor}`);
reg = await registrar.query(aor);
if (reg) {
// user is registered..find out which sbc is handling it
// us => we can put the call through
@@ -227,28 +158,34 @@ module.exports = (srf, logger, opts) => {
}
req.locals.registration = reg;
req.locals.target = 'user';
return next();
}
else {
// if the sip domain is one of ours return 404
const account = await lookupAccountBySipRealm(uri.host);
if (account) {
logger.info({host: uri.host, account}, `returning 404 to unregistered user in valid domain: ${req.uri}`);
res.send(404);
return;
}
else {
logger.info({host: uri.host, account}, `returning 404 to user in invalid domain: ${req.uri}`);
}
res.send(404);
return req.srf.endSession(req);
}
}
else if ('sip' === desiredRouting) {
if (!dotDecimalHost || !isLocalUri(uri.host, req)) {
// call that needs to be forwarded to a sip endpoint
logger.info(`forwarding call to sip endpoint ${req.uri}`);
req.locals.target = 'forward';
return next();
}
else if ('phone' === desiredRouting) {
debug('sending call to LCR');
req.locals.target = 'lcr';
// if the called number is digits only (after possible leading plus sign) and long enough, do lcr
if (!/^\d+$/.test(req.calledNumber.slice(1)) || req.calledNumber.length < 8) {
debug(`unable to route call to ${aor}; no registered user found`);
logger.info(`unable to route call to ${aor}; no registered user found`);
return res.send(404);
}
debug('sending call to LCR');
req.locals.target = 'lcr';
next();
};
+6 -113
View File
@@ -4,21 +4,12 @@ const debug = require('debug')('jambonz:sbc-outbound');
function makeRtpEngineOpts(req, srcIsUsingSrtp, dstIsUsingSrtp, teams = false) {
const from = req.getParsedHeader('from');
const rtpCopy = JSON.parse(JSON.stringify(rtpCharacteristics));
const srtpCopy = JSON.parse(JSON.stringify(srtpCharacteristics));
const srtpOpts = teams ? srtpCopy['teams'] : srtpCopy['default'];
const dstOpts = dstIsUsingSrtp ? srtpOpts : rtpCopy;
const srcOpts = srcIsUsingSrtp ? srtpOpts : rtpCopy;
/* webrtc clients (e.g. sipjs) send DMTF via SIP INFO */
if ((srcIsUsingSrtp || dstIsUsingSrtp) && !teams) {
dstOpts.flags.push('inject DTMF');
srcOpts.flags.push('inject DTMF');
}
const srtpOpts = teams ? srtpCharacteristics['teams'] : srtpCharacteristics['default'];
const dstOpts = dstIsUsingSrtp ? srtpOpts : rtpCharacteristics;
const srcOpts = srcIsUsingSrtp ? srtpOpts : rtpCharacteristics;
const common = {
'call-id': req.get('Call-ID'),
'replace': ['origin', 'session-connection'],
'record call': process.env.JAMBONES_RECORD_ALL_CALLS ? 'yes' : 'no'
'replace': ['origin', 'session-connection']
};
return {
common,
@@ -80,109 +71,11 @@ const pingMsTeamsGateways = (logger, srf) => {
});
};
const makeAccountCallCountKey = (sid) => `outcalls:account:${sid}`;
const makeSPCallCountKey = (sid) => `outcalls:sp:${sid}`;
const makeAppCallCountKey = (sid) => `outcalls:app:${sid}`;
const equalsIgnoreOrder = (a, b) => {
if (a.length !== b.length) return false;
const uniqueValues = new Set([...a, ...b]);
for (const v of uniqueValues) {
const aCount = a.filter((e) => e === v).length;
const bCount = b.filter((e) => e === v).length;
if (aCount !== bCount) return false;
}
return true;
};
const systemHealth = async(redisClient, ping, getCount) => {
await Promise.all([redisClient.ping(), ping()]);
return getCount();
};
const createHealthCheckApp = (port, logger) => {
const express = require('express');
const app = express();
app.use(express.urlencoded({ extended: true }));
app.use(express.json());
return new Promise((resolve) => {
app.listen(port, () => {
logger.info(`Health check server started at http://localhost:${port}`);
resolve(app);
});
});
};
const nudgeCallCounts = async(logger, sids, nudgeOperator, writers) => {
const {service_provider_sid, account_sid, application_sid} = sids;
const {writeCallCount, writeCallCountSP, writeCallCountApp} = writers;
const nudges = [];
const writes = [];
logger.debug(sids, 'nudgeCallCounts');
if (process.env.JAMBONES_TRACK_SP_CALLS) {
const key = makeSPCallCountKey(service_provider_sid);
nudges.push(nudgeOperator(key));
}
else {
nudges.push(() => Promise.resolve(null));
}
if (process.env.JAMBONES_TRACK_ACCOUNT_CALLS || process.env.JAMBONES_HOSTING) {
const key = makeAccountCallCountKey(account_sid);
nudges.push(nudgeOperator(key));
}
else {
nudges.push(() => Promise.resolve(null));
}
if (process.env.JAMBONES_TRACK_APP_CALLS && application_sid) {
const key = makeAppCallCountKey(application_sid);
nudges.push(nudgeOperator(key));
}
else {
nudges.push(() => Promise.resolve(null));
}
try {
const [callsSP, calls, callsApp] = await Promise.all(nudges);
logger.debug({
calls, callsSP, callsApp,
service_provider_sid, account_sid, application_sid}, 'call counts after adjustment');
if (process.env.JAMBONES_TRACK_SP_CALLS) {
writes.push(writeCallCountSP({service_provider_sid, calls_in_progress: callsSP}));
}
if (process.env.JAMBONES_TRACK_ACCOUNT_CALLS || process.env.JAMBONES_HOSTING) {
writes.push(writeCallCount({service_provider_sid, account_sid, calls_in_progress: calls}));
}
if (process.env.JAMBONES_TRACK_APP_CALLS && application_sid) {
writes.push(writeCallCountApp({service_provider_sid, account_sid, application_sid, calls_in_progress: callsApp}));
}
/* write the call counts to the database */
Promise.all(writes).catch((err) => logger.error({err}, 'Error writing call counts'));
return {callsSP, calls, callsApp};
} catch (err) {
logger.error(err, 'error incrementing call counts');
}
return {callsSP: null, calls: null, callsApp: null};
};
const makeCallCountKey = (sid) => `${sid}:outcalls`;
module.exports = {
makeRtpEngineOpts,
selectHostPort,
pingMsTeamsGateways,
makeAccountCallCountKey,
makeSPCallCountKey,
equalsIgnoreOrder,
systemHealth,
createHealthCheckApp,
nudgeCallCounts
makeCallCountKey
};
+2817 -4790
View File
File diff suppressed because it is too large Load Diff
+15 -20
View File
@@ -1,9 +1,9 @@
{
"name": "sbc-outbound",
"version": "v0.7.8",
"version": "0.4.6",
"main": "app.js",
"engines": {
"node": ">= 12.0.0"
"node": ">= 8.10.0"
},
"keywords": [
"sip",
@@ -22,32 +22,27 @@
"description": "jambonz session border controller application for outbound calls",
"scripts": {
"start": "node app",
"test": "NODE_ENV=test HTTP_PORT=3050 JAMBONES_HOSTING=1 JAMBONES_NETWORK_CIDR=127.0.0.1/32 JAMBONES_MYSQL_HOST=127.0.0.1 JAMBONES_MYSQL_USER=jambones_test JAMBONES_MYSQL_PASSWORD=jambones_test JAMBONES_MYSQL_DATABASE=jambones_test JAMBONES_REDIS_HOST=localhost JAMBONES_REDIS_PORT=16379 JAMBONES_TIME_SERIES_HOST=127.0.0.1 JAMBONES_LOGLEVEL=error DRACHTIO_SECRET=cymru DRACHTIO_HOST=127.0.0.1 DRACHTIO_PORT=9060 JAMBONES_RTPENGINES=127.0.0.1:12222 node test/ ",
"test": "NODE_ENV=test JAMBONZ_HOSTING=1 JAMBONES_NETWORK_CIDR=127.0.0.1/32 JAMBONES_MYSQL_HOST=127.0.0.1 JAMBONES_MYSQL_USER=jambones_test JAMBONES_MYSQL_PASSWORD=jambones_test JAMBONES_MYSQL_DATABASE=jambones_test JAMBONES_REDIS_HOST=localhost JAMBONES_REDIS_PORT=16379 JAMBONES_TIME_SERIES_HOST=127.0.0.1 JAMBONES_LOGLEVEL=error DRACHTIO_SECRET=cymru DRACHTIO_HOST=127.0.0.1 DRACHTIO_PORT=9060 JAMBONES_RTPENGINES=127.0.0.1:12222 node test/ ",
"coverage": "./node_modules/.bin/nyc --reporter html --report-dir ./coverage npm run test",
"jslint": "eslint app.js lib"
},
"dependencies": {
"@jambonz/db-helpers": "^0.7.3",
"@jambonz/realtimedb-helpers": "^0.4.35",
"@jambonz/http-health-check": "^0.0.1",
"@jambonz/mw-registrar": "0.2.2",
"@jambonz/rtpengine-utils": "^0.3.10",
"@jambonz/siprec-client-utils": "^0.1.4",
"@jambonz/stats-collector": "^0.1.6",
"@jambonz/time-series": "^0.2.5",
"@jambonz/db-helpers": "^0.6.12",
"@jambonz/mw-registrar": "0.2.1",
"@jambonz/realtimedb-helpers": "^0.4.3",
"@jambonz/rtpengine-utils": "^0.1.17",
"@jambonz/stats-collector": "^0.1.5",
"@jambonz/time-series": "^0.1.5",
"cidr-matcher": "^2.1.1",
"debug": "^4.3.4",
"debug": "^4.3.1",
"drachtio-fn-b2b-sugar": "^0.0.12",
"drachtio-srf": "^4.5.1",
"express": "^4.18.1",
"pino": "^7.11.0",
"sdp-transform": "^2.14.1"
"drachtio-srf": "^4.4.55",
"pino": "^6.11.2"
},
"devDependencies": {
"bent": "^7.3.12",
"eslint": "^7.32.0",
"eslint-plugin-promise": "^5.2.0",
"eslint": "^7.18.0",
"eslint-plugin-promise": "^4.2.1",
"nyc": "^15.1.0",
"tape": "^5.5.3"
"tape": "^5.2.2"
}
}
+1 -2
View File
@@ -10,7 +10,6 @@ networks:
services:
mysql:
image: mysql:5.7
platform: linux/x86_64
ports:
- "3306:3306"
environment:
@@ -119,7 +118,7 @@ services:
ipv4_address: 172.39.0.24
influxdb:
image: influxdb:1.8
image: influxdb:1.8-alpine
ports:
- "8086:8086"
networks:
-4
View File
@@ -5,7 +5,3 @@ DRACHTIO_SECRET=cymru
JAMBONES_REDIS_HOST=172.39.0.11
JAMBONES_REDIS_PORT=6379
JAMBONES_LOGLEVEL=info
JAMBONES_MYSQL_HOST=172.39.0.2
JAMBONES_MYSQL_USER=jambones_test
JAMBONES_MYSQL_PASSWORD=jambones_test
JAMBONES_MYSQL_DATABASE=jambones_test
+1 -1
View File
@@ -36,7 +36,6 @@
Subject: uac-pcap-carrier-success
X-Account-Sid: ed649e33-e771-403a-8c99-1780eabbc803
X-Call-Sid: fff49e33-e771-403a-8c99-1780eabbc803
X-Jambonz-Routing: phone
Content-Type: application/sdp
Content-Length: [len]
@@ -104,3 +103,4 @@
<CallLengthRepartition value="10, 50, 100, 500, 1000, 5000, 10000"/>
</scenario>
@@ -34,7 +34,6 @@
Contact: sip:sipp@[local_ip]:[local_port]
Max-Forwards: 70
Subject: uac-device-unknown-user
X-Jambonz-Routing: user
Content-Type: application/sdp
Content-Length: [len]
@@ -34,7 +34,6 @@
Contact: sip:sipp@[local_ip]:[local_port]
Max-Forwards: 70
Subject: uac-device-unknown-user
X-Jambonz-Routing: user
Content-Type: application/sdp
Content-Length: [len]
@@ -34,7 +34,6 @@
Contact: sip:sipp@[local_ip]:[local_port]
Max-Forwards: 70
Subject: uac-device-unknown-user
X-Jambonz-Routing: user
Content-Type: application/sdp
Content-Length: [len]
@@ -16,7 +16,6 @@
Max-Forwards: 70
X-Account-Sid: ed649e33-e771-403a-8c99-1780eabbc803
X-Call-Sid: ff49e33-e771-403a-8c99-1780eabbc803
X-Jambonz-Routing: phone
Subject: uac-pcap-carrier-fail-limits
Content-Type: application/sdp
Content-Length: [len]
@@ -17,7 +17,6 @@
Subject: uac-pcap-carrier-success
X-Account-Sid: ed649e33-e771-403a-8c99-1780eabbc803
X-Call-Sid: fff49e33-e771-403a-8c99-1780eabbc803
X-Jambonz-Routing: phone
Content-Type: application/sdp
Content-Length: [len]
@@ -16,7 +16,6 @@
Max-Forwards: 70
X-Account-Sid: ed649e33-e771-403a-8c99-1780eabbc803
X-Call-Sid: fff49e33-e771-403a-8c99-1780eabbc803
X-Jambonz-Routing: phone
Subject: uac-pcap-carrier-success
Content-Type: application/sdp
Content-Length: [len]
-1
View File
@@ -37,7 +37,6 @@
Subject: uac-pcap-device-success
X-Account-Sid: ed649e33-e771-403a-8c99-1780eabbc803
X-Call-Sid: ff649e33-e771-403a-8c99-1780eabbc803
X-Jambonz-Routing: user
Content-Type: application/sdp
Content-Length: [len]
@@ -34,7 +34,6 @@
Contact: sip:sipp@[local_ip]:[local_port]
Max-Forwards: 70
Subject: uac-pcap-device-success
X-Jambonz-Routing: user
Content-Type: application/sdp
Content-Length: [len]
@@ -35,7 +35,6 @@
Contact: sip:sipp@[local_ip]:[local_port]
Max-Forwards: 70
Subject: uac-pcap-device-success
X-Jambonz-Routing: user
Content-Type: application/sdp
Content-Length: [len]
+1 -3
View File
@@ -37,7 +37,6 @@
Subject: uac-sip-uri-auth-success
X-Account-Sid: ed649e33-e771-403a-8c99-1780eabbc803
X-Call-Sid: fff49e33-e771-403a-8c99-1780eabbc803
X-Jambonz-Routing: sip
Content-Type: application/sdp
Content-Length: [len]
@@ -91,7 +90,6 @@
Subject: uac-sip-uri-auth-success
X-Account-Sid: ed649e33-e771-403a-8c99-1780eabbc803
X-Call-Sid: fff49e33-e771-403a-8c99-1780eabbc803
X-Jambonz-Routing: sip
Content-Type: application/sdp
Content-Length: [len]
@@ -128,7 +126,7 @@
To: <sip:sip:john@jambonz.org>[peer_tag_param]
[last_Call-ID:]
CSeq: 2 ACK
Subject:
Subject: uac-sip-uri-auth-success
Content-Length: 0
]]>
+2 -8
View File
@@ -2,8 +2,7 @@ const test = require('tape');
const { output, sippUac } = require('./sipp')('test_sbc-outbound');
const {execSync} = require('child_process');
const debug = require('debug')('jambonz:sbc-outbound');
const bent = require('bent');
const getJSON = bent('json');
const consoleLogger = {error: console.error, info: console.log, debug: console.log};
process.on('unhandledRejection', (reason, p) => {
console.log('Unhandled Rejection at: Promise', p, 'reason:', reason);
@@ -30,11 +29,6 @@ test('sbc-outbound tests', async(t) => {
try {
await connect(srf);
let obj = await getJSON('http://127.0.0.1:3050/');
t.ok(obj.calls === 0, 'HTTP GET / works (current call count)')
obj = await getJSON('http://127.0.0.1:3050/system-health');
t.ok(obj.calls === 0, 'HTTP GET /system-health works (health check)')
/* call to unregistered user */
debug('successfully connected to drachtio server');
await sippUac('uac-pcap-device-404.xml');
@@ -84,7 +78,7 @@ test('sbc-outbound tests', async(t) => {
const res = await queryCdrs({account_sid: 'ed649e33-e771-403a-8c99-1780eabbc803'});
//console.log(`cdrs: ${JSON.stringify(res)}`);
t.ok(res.total === 5, 'wrote 5 cdrs');
t.ok(res.total === 6, 'wrote 6 cdrs');
srf.disconnect();
} catch (err) {