a number with no fallback URL drops the call when yours 500s

"""Report Twilio numbers whose live handlers have no fallback 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 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_fallback_audit")

HOST = "https://api.twilio.com"
BASE = HOST + "/2010-04-01"

CHANNELS = (
    ("voice", "voice_url", "voice_fallback_url", "voice_application_sid"),
    ("sms", "sms_url", "sms_fallback_url", "sms_application_sid"),
)


def verdict(number, apps=None):
    """Classify one IncomingPhoneNumber. Pure, so the precedence rule can be
    tested without a network.

    `apps` maps an Application SID to that Application resource. When a channel
    has an application sid, the application is the effective handler and the
    number's own url and fallback are ignored entirely.

    Returns (state, detail).
    """
    apps = apps or {}
    exposed, covered, unresolved = [], [], []

    for channel, url_field, fb_field, app_field in CHANNELS:
        app_sid = str(number.get(app_field) or "").strip()
        if app_sid:
            source = apps.get(app_sid)
            if source is None:
                unresolved.append("%s (%s)" % (channel, app_sid))
                continue
            where = "app %s" % app_sid
        else:
            source, where = number, "the number"
        primary = str(source.get(url_field) or "").strip()
        fallback = str(source.get(fb_field) or "").strip()
        if not primary:
            continue
        (covered if fallback else exposed).append("%s on %s" % (channel, where))

    if unresolved:
        return ("unresolved",
                "an application sid is set but the application was not read: %s"
                % ", ".join(unresolved))
    if exposed:
        return ("exposed",
                "%s has a live handler and no fallback: one non-2xx and the "
                "interaction is dropped." % "; ".join(exposed))
    if covered:
        return ("covered", "fallback set for " + ", ".join(covered))
    return ("idle", "no voice or sms handler configured, so nothing to fall back from")


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_numbers(session, account, limit):
    url = "%s/Accounts/%s/IncomingPhoneNumbers.json" % (BASE, account)
    params = {"PageSize": 100}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("incoming_phone_numbers", []))
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def load_apps(session, account, numbers):
    """Fetch each referenced Application once. A busy account points many numbers
    at the same app, so this is a handful of GETs rather than one per number."""
    sids = set()
    for n in numbers:
        for _c, _u, _f, app_field in CHANNELS:
            sid = str(n.get(app_field) or "").strip()
            if sid:
                sids.add(sid)
    apps = {}
    for sid in sorted(sids):
        apps[sid] = get(session, "%s/Accounts/%s/Applications/%s.json"
                        % (BASE, account, sid))
    return apps


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-numbers", type=int, default=1000)
    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)

    numbers = list_numbers(session, account, args.max_numbers)
    if not numbers:
        log.info("no phone numbers on this account")
        return 0
    apps = load_apps(session, account, numbers)

    bad = 0
    for n in numbers:
        state, detail = verdict(n, apps)
        line = "%-10s %s  %s" % (state, n.get("phone_number", "?"), detail)
        if state in ("covered", "idle"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        log.warning("  repair: POST %s/Accounts/%s/IncomingPhoneNumbers/%s.json "
                    "VoiceFallbackUrl=https://handler.twilio.com/twiml/EHxxx "
                    "VoiceFallbackMethod=POST", BASE, account, n.get("sid"))

    log.info("%d number(s), %d with an unprotected handler", len(numbers), bad)
    return 1 if bad else 0


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

/**
 * Report Twilio numbers whose live handlers have no fallback 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 HOST = 'https://api.twilio.com';
const BASE = `${HOST}/2010-04-01`;

const CHANNELS = [
  ['voice', 'voice_url', 'voice_fallback_url', 'voice_application_sid'],
  ['sms', 'sms_url', 'sms_fallback_url', 'sms_application_sid'],
];

/**
 * Classify one IncomingPhoneNumber. Pure, so the precedence rule can be tested
 * without a network. `apps` maps an Application SID to that Application: when a
 * channel has one, it is the effective handler and the number's own url and
 * fallback are ignored entirely. Returns [state, detail].
 */
export function verdict(number, apps = {}) {
  const exposed = [];
  const covered = [];
  const unresolved = [];

  for (const [channel, urlField, fbField, appField] of CHANNELS) {
    const appSid = String(number[appField] ?? '').trim();
    let source = number;
    let where = 'the number';
    if (appSid) {
      source = apps[appSid];
      if (source === undefined) { unresolved.push(`${channel} (${appSid})`); continue; }
      where = `app ${appSid}`;
    }
    const primary = String(source[urlField] ?? '').trim();
    const fallback = String(source[fbField] ?? '').trim();
    if (!primary) continue;
    (fallback ? covered : exposed).push(`${channel} on ${where}`);
  }

  if (unresolved.length) {
    return ['unresolved',
      `an application sid is set but the application was not read: ${unresolved.join(', ')}`];
  }
  if (exposed.length) {
    return ['exposed',
      `${exposed.join('; ')} has a live handler and no fallback: one non-2xx ` +
      'and the interaction is dropped.'];
  }
  if (covered.length) return ['covered', `fallback set for ${covered.join(', ')}`];
  return ['idle', 'no voice or sms handler configured, so nothing to fall back from'];
}

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();
}

export async function listNumbers(auth, account, limit = 1000) {
  let url = `${BASE}/Accounts/${account}/IncomingPhoneNumbers.json`;
  let params = { PageSize: 100 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.incoming_phone_numbers ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

async function loadApps(auth, account, numbers) {
  const sids = new Set();
  for (const n of numbers) {
    for (const [, , , appField] of CHANNELS) {
      const sid = String(n[appField] ?? '').trim();
      if (sid) sids.add(sid);
    }
  }
  const apps = {};
  for (const sid of [...sids].sort()) {
    apps[sid] = await get(auth, `${BASE}/Accounts/${account}/Applications/${sid}.json`);
  }
  return apps;
}

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 numbers = await listNumbers(auth, account);
  if (numbers.length === 0) {
    console.log('no phone numbers on this account');
    return;
  }
  const apps = await loadApps(auth, account, numbers);

  let bad = 0;
  for (const n of numbers) {
    const [state, detail] = verdict(n, apps);
    const line = `${state.padEnd(10)} ${n.phone_number ?? '?'}  ${detail}`;
    if (state === 'covered' || state === 'idle') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    console.warn(`  repair: POST ${BASE}/Accounts/${account}/IncomingPhoneNumbers/` +
                 `${n.sid}.json VoiceFallbackUrl=https://handler.twilio.com/twiml/EHxxx ` +
                 'VoiceFallbackMethod=POST');
  }

  console.log(`${numbers.length} number(s), ${bad} with an unprotected handler`);
  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_fallback_audit import verdict

APP = "AP0123456789"


def test_live_voice_handler_with_no_fallback_is_exposed():
    state, detail = verdict({"voice_url": "https://app.example.com/voice"})
    assert state == "exposed"
    assert "dropped" in detail


def test_fallback_on_the_number_is_covered():
    state, _ = verdict({"voice_url": "https://app.example.com/voice",
                        "voice_fallback_url": "https://handler.twilio.com/twiml/EH1"})
    assert state == "covered"


def test_application_sid_wins_so_a_fallback_on_the_number_does_not_count():
    # The mistake this note exists to prevent: the number looks protected and is not.
    state, detail = verdict(
        {"voice_application_sid": APP,
         "voice_url": "https://app.example.com/voice",
         "voice_fallback_url": "https://handler.twilio.com/twiml/EH1"},
        {APP: {"voice_url": "https://app.example.com/voice"}})
    assert state == "exposed"
    assert APP in detail


def test_fallback_on_the_application_counts():
    state, _ = verdict(
        {"voice_application_sid": APP},
        {APP: {"voice_url": "https://app.example.com/voice",
               "voice_fallback_url": "https://handler.twilio.com/twiml/EH1"}})
    assert state == "covered"


def test_sms_is_checked_when_voice_is_fine():
    state, detail = verdict({"voice_url": "https://app.example.com/voice",
                             "voice_fallback_url": "https://handler.twilio.com/twiml/EH1",
                             "sms_url": "https://app.example.com/sms"})
    assert state == "exposed"
    assert "sms" in detail


def test_number_with_no_handler_is_idle_not_exposed():
    state, _ = verdict({"voice_url": "", "sms_url": None})
    assert state == "idle"


def test_unread_application_is_not_guessed_at():
    state, _ = verdict({"voice_application_sid": APP}, {})
    assert state == "unresolved"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict } from './twilio-fallback-audit.mjs';

const APP = 'AP0123456789';

test('live voice handler with no fallback is exposed', () => {
  const [state, detail] = verdict({ voice_url: 'https://app.example.com/voice' });
  assert.equal(state, 'exposed');
  assert.match(detail, /dropped/);
});

test('fallback on the number is covered', () => {
  const [state] = verdict({
    voice_url: 'https://app.example.com/voice',
    voice_fallback_url: 'https://handler.twilio.com/twiml/EH1',
  });
  assert.equal(state, 'covered');
});

test('application sid wins, so a fallback on the number does not count', () => {
  const [state, detail] = verdict(
    { voice_application_sid: APP,
      voice_url: 'https://app.example.com/voice',
      voice_fallback_url: 'https://handler.twilio.com/twiml/EH1' },
    { [APP]: { voice_url: 'https://app.example.com/voice' } });
  assert.equal(state, 'exposed');
  assert.match(detail, new RegExp(APP));
});

test('fallback on the application counts', () => {
  const [state] = verdict(
    { voice_application_sid: APP },
    { [APP]: { voice_url: 'https://app.example.com/voice',
               voice_fallback_url: 'https://handler.twilio.com/twiml/EH1' } });
  assert.equal(state, 'covered');
});

test('sms is checked when voice is fine', () => {
  const [state, detail] = verdict({
    voice_url: 'https://app.example.com/voice',
    voice_fallback_url: 'https://handler.twilio.com/twiml/EH1',
    sms_url: 'https://app.example.com/sms',
  });
  assert.equal(state, 'exposed');
  assert.match(detail, /sms/);
});

test('number with no handler is idle, not exposed', () => {
  assert.equal(verdict({ voice_url: '', sms_url: null })[0], 'idle');
});

test('unread application is not guessed at', () => {
  assert.equal(verdict({ voice_application_sid: APP }, {})[0], 'unresolved');
});