a phone number still points at Twilio's demo TwiML

"""Report Twilio phone numbers still answering with demo or placeholder TwiML.

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 send messages 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_demo_twiml_audit")

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

DEMO_HOST = "demo.twilio.com"
BIN_PREFIX = "handler.twilio.com/twiml/"


def host_and_path(url):
    """Reduce a URL to lowercase host plus path.

    The demo endpoint turns up as http and https, with and without a query
    string, and pointing at several different demo documents. Matching the whole
    string misses most of those; matching host and path catches all of them.
    """
    u = str(url or "").strip()
    for scheme in ("https://", "http://"):
        if u.lower().startswith(scheme):
            u = u[len(scheme):]
            break
    u = u.split("?", 1)[0].split("#", 1)[0]
    head = u.split("/", 1)[0]
    if "@" in head:
        u = u.split("@", 1)[1]
    return u.lower()


def verdict(number):
    """Classify one IncomingPhoneNumber. Pure, so the rules can be tested
    without a network.

    Returns (state, detail).
    """
    handlers = [("voice", number.get("voice_url")), ("sms", number.get("sms_url"))]

    demo = [c for c, u in handlers if host_and_path(u).startswith(DEMO_HOST)]
    if demo:
        return ("demo",
                "%s handler is Twilio's demo TwiML. It answers 200 with valid "
                "TwiML, so nothing is logged and every call reads as completed."
                % "/".join(demo))

    bins = [c for c, u in handlers if host_and_path(u).startswith(BIN_PREFIX)]
    if bins:
        return ("twiml-bin",
                "%s handler is a TwiML Bin. Bins are legitimate, but one left "
                "over from a quickstart fails exactly like the demo URL."
                % "/".join(bins))

    routed = [c for c, u in handlers if str(u or "").strip()]
    if str(number.get("voice_application_sid") or "").strip():
        routed.append("voice app")
    if str(number.get("sms_application_sid") or "").strip():
        routed.append("sms app")
    if not routed:
        return ("unrouted",
                "no voice_url, no sms_url and no application sid: the number is "
                "bought, billed monthly and answers nothing.")

    return ("configured", "handled by " + ", ".join(routed))


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):
    """Page IncomingPhoneNumbers. next_page_uri is a path, not an absolute URL."""
    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 has_traffic(session, account, e164):
    """One call record is enough to know the number is in use."""
    page = get(session, "%s/Accounts/%s/Calls.json" % (BASE, account),
               To=e164, PageSize=1)
    return bool(page.get("calls"))


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-numbers", type=int, default=1000,
                    help="stop after this many numbers")
    ap.add_argument("--check-traffic", action="store_true",
                    help="one extra GET per flagged number to see if it is dialled")
    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

    bad = 0
    for n in numbers:
        state, detail = verdict(n)
        line = "%-11s %s  %s" % (state, n.get("phone_number", "?"), detail)
        if state == "configured":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if args.check_traffic and has_traffic(session, account, n.get("phone_number")):
            log.warning("  this number has inbound calls: fix it before the rest")
        log.warning("  repair: POST %s/Accounts/%s/IncomingPhoneNumbers/%s.json "
                    "VoiceUrl=https://your-app.example.com/voice VoiceMethod=POST",
                    BASE, account, n.get("sid"))

    log.info("%d number(s), %d on demo or placeholder TwiML", len(numbers), bad)
    return 1 if bad else 0


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

/**
 * Report Twilio phone numbers still answering with demo or placeholder TwiML.
 *
 * 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 DEMO_HOST = 'demo.twilio.com';
const BIN_PREFIX = 'handler.twilio.com/twiml/';

/**
 * Reduce a URL to lowercase host plus path, so http/https, query strings and
 * different demo documents all match the same rule.
 */
export function hostAndPath(url) {
  let u = String(url ?? '').trim();
  for (const scheme of ['https://', 'http://']) {
    if (u.toLowerCase().startsWith(scheme)) { u = u.slice(scheme.length); break; }
  }
  u = u.split('?')[0].split('#')[0];
  if (u.split('/')[0].includes('@')) u = u.slice(u.indexOf('@') + 1);
  return u.toLowerCase();
}

/**
 * Classify one IncomingPhoneNumber. Pure, so the rules can be tested without a
 * network. Returns [state, detail].
 */
export function verdict(number) {
  const handlers = [['voice', number.voice_url], ['sms', number.sms_url]];

  const demo = handlers.filter(([, u]) => hostAndPath(u).startsWith(DEMO_HOST));
  if (demo.length) {
    return ['demo',
      `${demo.map(([c]) => c).join('/')} handler is Twilio's demo TwiML. It ` +
      'answers 200 with valid TwiML, so nothing is logged and every call reads ' +
      'as completed.'];
  }

  const bins = handlers.filter(([, u]) => hostAndPath(u).startsWith(BIN_PREFIX));
  if (bins.length) {
    return ['twiml-bin',
      `${bins.map(([c]) => c).join('/')} handler is a TwiML Bin. Bins are ` +
      'legitimate, but one left over from a quickstart fails exactly like the ' +
      'demo URL.'];
  }

  const routed = handlers.filter(([, u]) => String(u ?? '').trim()).map(([c]) => c);
  if (String(number.voice_application_sid ?? '').trim()) routed.push('voice app');
  if (String(number.sms_application_sid ?? '').trim()) routed.push('sms app');
  if (routed.length === 0) {
    return ['unrouted',
      'no voice_url, no sms_url and no application sid: the number is bought, ' +
      'billed monthly and answers nothing.'];
  }

  return ['configured', `handled by ${routed.join(', ')}`];
}

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 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 checkTraffic = process.argv.includes('--check-traffic');

  const numbers = await listNumbers(auth, account);
  if (numbers.length === 0) {
    console.log('no phone numbers on this account');
    return;
  }

  let bad = 0;
  for (const n of numbers) {
    const [state, detail] = verdict(n);
    const line = `${state.padEnd(11)} ${n.phone_number ?? '?'}  ${detail}`;
    if (state === 'configured') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    if (checkTraffic) {
      const calls = await get(auth, `${BASE}/Accounts/${account}/Calls.json`,
                              { To: n.phone_number, PageSize: 1 });
      if ((calls.calls ?? []).length) {
        console.warn('  this number has inbound calls: fix it before the rest');
      }
    }
    console.warn(`  repair: POST ${BASE}/Accounts/${account}/IncomingPhoneNumbers/` +
                 `${n.sid}.json VoiceUrl=https://your-app.example.com/voice ` +
                 'VoiceMethod=POST');
  }

  console.log(`${numbers.length} number(s), ${bad} on demo or placeholder TwiML`);
  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_demo_twiml_audit import host_and_path, verdict


def test_default_demo_voice_url_is_flagged():
    state, detail = verdict({"voice_url": "https://demo.twilio.com/docs/voice.xml"})
    assert state == "demo"
    assert "completed" in detail


def test_demo_url_over_http_and_with_a_query_string_is_still_demo():
    # The reason matching is on host and path rather than the whole string.
    state, _ = verdict({"voice_url": "http://demo.twilio.com/docs/voice.xml?x=1"})
    assert state == "demo"


def test_demo_on_the_sms_handler_is_found_when_voice_is_fine():
    state, detail = verdict({"voice_url": "https://app.example.com/voice",
                             "sms_url": "https://demo.twilio.com/welcome/sms/reply"})
    assert state == "demo"
    assert "sms" in detail


def test_unedited_twiml_bin_is_its_own_state():
    state, _ = verdict({"voice_url": "https://handler.twilio.com/twiml/EH0123456789"})
    assert state == "twiml-bin"


def test_number_with_no_handler_at_all_is_unrouted():
    state, detail = verdict({"voice_url": "", "sms_url": None})
    assert state == "unrouted"
    assert "billed" in detail


def test_application_sid_counts_as_routed():
    state, _ = verdict({"voice_application_sid": "AP0123456789"})
    assert state == "configured"


def test_host_and_path_drops_scheme_credentials_and_query():
    assert host_and_path("https://user@Demo.Twilio.com/docs/voice.xml?a=b") == \
        "demo.twilio.com/docs/voice.xml"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { hostAndPath, verdict } from './twilio-demo-twiml-audit.mjs';

test('default demo voice url is flagged', () => {
  const [state, detail] = verdict({ voice_url: 'https://demo.twilio.com/docs/voice.xml' });
  assert.equal(state, 'demo');
  assert.match(detail, /completed/);
});

test('demo url over http and with a query string is still demo', () => {
  assert.equal(verdict({ voice_url: 'http://demo.twilio.com/docs/voice.xml?x=1' })[0], 'demo');
});

test('demo on the sms handler is found when voice is fine', () => {
  const [state, detail] = verdict({
    voice_url: 'https://app.example.com/voice',
    sms_url: 'https://demo.twilio.com/welcome/sms/reply',
  });
  assert.equal(state, 'demo');
  assert.match(detail, /sms/);
});

test('unedited twiml bin is its own state', () => {
  assert.equal(
    verdict({ voice_url: 'https://handler.twilio.com/twiml/EH0123456789' })[0],
    'twiml-bin');
});

test('number with no handler at all is unrouted', () => {
  const [state, detail] = verdict({ voice_url: '', sms_url: null });
  assert.equal(state, 'unrouted');
  assert.match(detail, /billed/);
});

test('application sid counts as routed', () => {
  assert.equal(verdict({ voice_application_sid: 'AP0123456789' })[0], 'configured');
});

test('hostAndPath drops scheme, credentials and query', () => {
  assert.equal(hostAndPath('https://user@Demo.Twilio.com/docs/voice.xml?a=b'),
               'demo.twilio.com/docs/voice.xml');
});