a SIP trunk with no disaster recovery URL loses every call

"""Report Twilio SIP Trunks with no disaster recovery URL.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The repair is printed, never performed,
because this script holds a credential to an account that can place calls and
spend money.
"""
import argparse
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("twilio_trunk_dr_audit")

TRUNKING = "https://trunking.twilio.com/v1"


def scheme_of(url):
    """Lowercase URL scheme, or an empty string when there is not one.

    Kept separate because a disaster recovery URL on plain http is a different
    finding from one that is missing, and the difference is one substring.
    """
    u = str(url or "").strip()
    if "://" not in u:
        return ""
    return u.split("://", 1)[0].lower()


def enabled_uris(origination):
    """The origination URIs Twilio would actually try.

    A disabled URI is still in the listing and still has a sip_url, so counting
    the list rather than the enabled subset overstates the redundancy exactly
    when it matters.
    """
    return [u for u in (origination or []) if u.get("enabled")]


def verdict(trunk, origination=None):
    """Classify one Trunk. Pure, so the rules can be tested without a network.

    origination is the trunk's OriginationUrl list, or None when it was not
    fetched. None and an empty list mean different things: the first is "not
    checked", the second is "checked, and there is nowhere for calls to go".

    Returns (state, detail).
    """
    dr = str(trunk.get("disaster_recovery_url") or "").strip()
    if not dr:
        return ("exposed",
                "no disaster_recovery_url: when the origination URIs stop "
                "answering, inbound calls to this trunk end at Twilio with no "
                "fallback, no voicemail and nothing logged as a call failure.")

    if scheme_of(dr) == "http":
        return ("dr-cleartext",
                "disaster_recovery_url is plain http, so the one TwiML fetch "
                "that happens while your voice path is already degraded crosses "
                "the public internet in cleartext.")

    if origination is not None:
        live = enabled_uris(origination)
        if not live:
            return ("no-origination",
                    "disaster recovery is set, but no origination URI is "
                    "enabled: inbound calls have nowhere to go on a good day, "
                    "not only during an outage.")
        if len(live) == 1:
            return ("single-uri",
                    "one enabled origination URI (%s), so the disaster recovery "
                    "URL is the only cover for that single host."
                    % (live[0].get("sip_url") or "?"))

    method = str(trunk.get("disaster_recovery_method") or "").strip().upper()
    return ("covered",
            "disaster_recovery_url is set and will be fetched with %s"
            % (method or "the default, which is a POST"))


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Twilio: check TWILIO_ACCOUNT_SID and that the "
                         "API key belongs to that account with read access"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def list_trunks(session, limit):
    """Page the trunks. This API paginates with an absolute meta.next_page_url."""
    url = TRUNKING + "/Trunks"
    params = {"PageSize": 100}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("trunks", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def list_origination(session, trunk_sid):
    """Origination URIs for one trunk. Not paginated in practice, but read the
    meta anyway rather than assuming."""
    url = "%s/Trunks/%s/OriginationUrls" % (TRUNKING, trunk_sid)
    params = {"PageSize": 100}
    out = []
    while url:
        page = get(session, url, **params)
        out.extend(page.get("origination_urls", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-trunks", type=int, default=200,
                    help="stop after this many trunks")
    ap.add_argument("--check-origination", action="store_true",
                    help="one extra GET per trunk to count enabled origination URIs")
    args = ap.parse_args()

    account = os.environ.get("TWILIO_ACCOUNT_SID")
    key = os.environ.get("TWILIO_API_KEY")
    secret = os.environ.get("TWILIO_API_SECRET")
    if not (account and key and secret):
        log.error("set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET "
                  "(an API Key with read access, not the auth token)")
        return 2

    session = requests.Session()
    session.auth = (key, secret)

    trunks = list_trunks(session, args.max_trunks)
    if not trunks:
        log.info("no SIP trunks on this account")
        return 0

    bad = 0
    for t in trunks:
        origination = None
        if args.check_origination:
            origination = list_origination(session, t.get("sid"))
        state, detail = verdict(t, origination)
        name = t.get("friendly_name") or t.get("domain_name") or t.get("sid")
        line = "%-14s %s  %s" % (state, name, detail)
        if state == "covered":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        log.warning("  secure=%s transfer_mode=%s",
                    t.get("secure"), t.get("transfer_mode"))
        log.warning("  repair: POST %s/Trunks/%s "
                    "DisasterRecoveryUrl=https://your-app.example.com/dr-twiml "
                    "DisasterRecoveryMethod=POST", TRUNKING, t.get("sid"))
        log.warning("  host that TwiML somewhere that does not depend on the PBX")

    log.info("%d trunk(s), %d without disaster recovery", len(trunks), bad)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report Twilio SIP Trunks with no disaster recovery URL.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The repair is printed, never
 * performed.
 */
const TRUNKING = 'https://trunking.twilio.com/v1';

/**
 * Lowercase URL scheme, or an empty string when there is not one. A disaster
 * recovery URL on plain http is a different finding from a missing one.
 */
export function schemeOf(url) {
  const u = String(url ?? '').trim();
  if (!u.includes('://')) return '';
  return u.split('://')[0].toLowerCase();
}

/**
 * The origination URIs Twilio would actually try. A disabled URI is still in
 * the listing, so counting the list overstates the redundancy.
 */
export function enabledUris(origination) {
  return (origination ?? []).filter((u) => u.enabled);
}

/**
 * Classify one Trunk. Pure, so the rules can be tested without a network.
 * `origination` is the trunk's OriginationUrl list, or null when it was not
 * fetched: null means "not checked", an empty array means "checked, and there
 * is nowhere for calls to go". Returns [state, detail].
 */
export function verdict(trunk, origination = null) {
  const dr = String(trunk.disaster_recovery_url ?? '').trim();
  if (!dr) {
    return ['exposed',
      'no disaster_recovery_url: when the origination URIs stop answering, ' +
      'inbound calls to this trunk end at Twilio with no fallback, no ' +
      'voicemail and nothing logged as a call failure.'];
  }

  if (schemeOf(dr) === 'http') {
    return ['dr-cleartext',
      'disaster_recovery_url is plain http, so the one TwiML fetch that ' +
      'happens while your voice path is already degraded crosses the public ' +
      'internet in cleartext.'];
  }

  if (origination !== null) {
    const live = enabledUris(origination);
    if (live.length === 0) {
      return ['no-origination',
        'disaster recovery is set, but no origination URI is enabled: inbound ' +
        'calls have nowhere to go on a good day, not only during an outage.'];
    }
    if (live.length === 1) {
      return ['single-uri',
        `one enabled origination URI (${live[0].sip_url ?? '?'}), so the ` +
        'disaster recovery URL is the only cover for that single host.'];
    }
  }

  const method = String(trunk.disaster_recovery_method ?? '').trim().toUpperCase();
  return ['covered',
    `disaster_recovery_url is set and will be fetched with ${method || 'the default, which is a POST'}`];
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

async function get(auth, url, params = {}) {
  const u = new URL(url);
  for (const [k, v] of Object.entries(params)) u.searchParams.set(k, v);
  const res = await fetch(u, { headers: { Authorization: auth } });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Twilio: check TWILIO_ACCOUNT_SID and ` +
                    'that the API key belongs to that account with read access');
  }
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return res.json();
}

/** Page the trunks. This API paginates with an absolute meta.next_page_url. */
export async function listTrunks(auth, limit = 200) {
  let url = `${TRUNKING}/Trunks`;
  let params = { PageSize: 100 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.trunks ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

export async function listOrigination(auth, trunkSid) {
  let url = `${TRUNKING}/Trunks/${trunkSid}/OriginationUrls`;
  let params = { PageSize: 100 };
  const out = [];
  while (url) {
    const page = await get(auth, url, params);
    out.push(...(page.origination_urls ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out;
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;
  if (!account || !key || !secret) {
    console.error('set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET ' +
                  '(an API Key with read access, not the auth token)');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const checkOrigination = process.argv.includes('--check-origination');

  const trunks = await listTrunks(auth);
  if (trunks.length === 0) {
    console.log('no SIP trunks on this account');
    return;
  }

  let bad = 0;
  for (const t of trunks) {
    const origination = checkOrigination ? await listOrigination(auth, t.sid) : null;
    const [state, detail] = verdict(t, origination);
    const name = t.friendly_name || t.domain_name || t.sid;
    const line = `${state.padEnd(14)} ${name}  ${detail}`;
    if (state === 'covered') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    console.warn(`  secure=${t.secure} transfer_mode=${t.transfer_mode}`);
    console.warn(`  repair: POST ${TRUNKING}/Trunks/${t.sid} ` +
                 'DisasterRecoveryUrl=https://your-app.example.com/dr-twiml ' +
                 'DisasterRecoveryMethod=POST');
    console.warn('  host that TwiML somewhere that does not depend on the PBX');
  }

  console.log(`${trunks.length} trunk(s), ${bad} without disaster recovery`);
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing credentials and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from twilio_trunk_dr_audit import enabled_uris, scheme_of, verdict


def test_empty_disaster_recovery_url_is_exposed():
    state, detail = verdict({"disaster_recovery_url": ""})
    assert state == "exposed"
    assert "no fallback" in detail


def test_missing_field_reads_the_same_as_an_empty_one():
    assert verdict({})[0] == "exposed"
    assert verdict({"disaster_recovery_url": None})[0] == "exposed"


def test_method_without_a_url_is_still_exposed():
    # A method is not a destination. Reading the pair as configured because one
    # half is populated is the mistake this case exists to prevent.
    assert verdict({"disaster_recovery_method": "POST"})[0] == "exposed"


def test_cleartext_disaster_recovery_url_is_its_own_state():
    state, _ = verdict({"disaster_recovery_url": "http://dr.example.com/twiml"})
    assert state == "dr-cleartext"


def test_https_url_with_no_origination_check_is_covered():
    # origination=None means not checked, and must not be read as "no URIs".
    state, detail = verdict({"disaster_recovery_url": "https://dr.example.com/twiml"})
    assert state == "covered"
    assert "the default" in detail


def test_checked_and_empty_origination_is_not_the_same_as_unchecked():
    state, _ = verdict({"disaster_recovery_url": "https://dr.example.com/twiml"}, [])
    assert state == "no-origination"


def test_disabled_uris_do_not_count_towards_redundancy():
    origination = [
        {"sip_url": "sip:a.example.com", "enabled": True},
        {"sip_url": "sip:b.example.com", "enabled": False},
        {"sip_url": "sip:c.example.com", "enabled": False},
    ]
    state, detail = verdict(
        {"disaster_recovery_url": "https://dr.example.com/twiml"}, origination)
    assert state == "single-uri"
    assert "a.example.com" in detail
    assert len(enabled_uris(origination)) == 1


def test_two_live_uris_and_a_recovery_url_is_covered():
    origination = [{"sip_url": "sip:a", "enabled": True},
                   {"sip_url": "sip:b", "enabled": True}]
    assert verdict({"disaster_recovery_url": "https://dr.example.com/twiml",
                    "disaster_recovery_method": "post"}, origination)[0] == "covered"


def test_scheme_of_handles_a_bare_host():
    assert scheme_of("HTTPS://dr.example.com/x") == "https"
    assert scheme_of("dr.example.com/x") == ""

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { enabledUris, schemeOf, verdict } from './twilio-trunk-dr-audit.mjs';

test('empty disaster recovery url is exposed', () => {
  const [state, detail] = verdict({ disaster_recovery_url: '' });
  assert.equal(state, 'exposed');
  assert.match(detail, /no fallback/);
});

test('missing field reads the same as an empty one', () => {
  assert.equal(verdict({})[0], 'exposed');
  assert.equal(verdict({ disaster_recovery_url: null })[0], 'exposed');
});

test('method without a url is still exposed', () => {
  assert.equal(verdict({ disaster_recovery_method: 'POST' })[0], 'exposed');
});

test('cleartext disaster recovery url is its own state', () => {
  assert.equal(verdict({ disaster_recovery_url: 'http://dr.example.com/twiml' })[0],
               'dr-cleartext');
});

test('https url with no origination check is covered', () => {
  const [state, detail] = verdict({ disaster_recovery_url: 'https://dr.example.com/twiml' });
  assert.equal(state, 'covered');
  assert.match(detail, /the default/);
});

test('checked and empty origination is not the same as unchecked', () => {
  assert.equal(
    verdict({ disaster_recovery_url: 'https://dr.example.com/twiml' }, [])[0],
    'no-origination');
});

test('disabled uris do not count towards redundancy', () => {
  const origination = [
    { sip_url: 'sip:a.example.com', enabled: true },
    { sip_url: 'sip:b.example.com', enabled: false },
    { sip_url: 'sip:c.example.com', enabled: false },
  ];
  const [state, detail] = verdict(
    { disaster_recovery_url: 'https://dr.example.com/twiml' }, origination);
  assert.equal(state, 'single-uri');
  assert.match(detail, /a\.example\.com/);
  assert.equal(enabledUris(origination).length, 1);
});

test('two live uris and a recovery url is covered', () => {
  const origination = [{ sip_url: 'sip:a', enabled: true },
                       { sip_url: 'sip:b', enabled: true }];
  assert.equal(verdict({ disaster_recovery_url: 'https://dr.example.com/twiml',
                         disaster_recovery_method: 'post' }, origination)[0], 'covered');
});

test('schemeOf handles a bare host', () => {
  assert.equal(schemeOf('HTTPS://dr.example.com/x'), 'https');
  assert.equal(schemeOf('dr.example.com/x'), '');
});