a number with an Application SID ignores its own voice_url

"""Report Twilio numbers whose webhook URLs are shadowed by an Application SID.

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_number_app_precedence_audit")

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

# (channel, url field, application sid field). The Application resource happens
# to name its URLs identically, which is what lets one comparison serve both.
CHANNELS = (
    ("voice", "voice_url", "voice_application_sid"),
    ("sms", "sms_url", "sms_application_sid"),
)


def verdict(number, apps=None):
    """Classify one IncomingPhoneNumber against the apps it references.

    Pure, so the precedence rule is testable without a network. `apps` maps an
    Application SID to that Application resource. When a channel carries an
    application sid, the Application is the effective handler and the number's
    own url is never requested.

    Returns (state, detail).
    """
    apps = apps or {}
    unresolved, dead, shadowed, routed, direct = [], [], [], [], []

    for channel, url_field, app_field in CHANNELS:
        app_sid = str(number.get(app_field) or "").strip()
        own = str(number.get(url_field) or "").strip()

        if not app_sid:
            if own:
                direct.append("%s serves %s" % (channel, own))
            continue

        app = apps.get(app_sid)
        if app is None:
            unresolved.append("%s (%s)" % (channel, app_sid))
            continue

        live = str(app.get(url_field) or "").strip()
        if not live:
            dead.append("%s: app %s has no %s" % (channel, app_sid, url_field))
            continue
        if own and own != live:
            shadowed.append("%s: %s on the number is ignored, app %s serves %s"
                            % (channel, own, app_sid, live))
            continue
        routed.append("%s via app %s" % (channel, app_sid))

    if unresolved:
        return ("unresolved",
                "an application sid is set but that application was not read: %s"
                % ", ".join(unresolved))
    if dead:
        return ("routes-nowhere",
                "%s. The number's own url cannot rescue this: the app wins while "
                "it is attached." % "; ".join(dead))
    if shadowed:
        return ("shadowed",
                "%s. Editing the number changes nothing." % "; ".join(shadowed))
    if routed:
        return ("app-routed", "handled by its application: " + ", ".join(routed))
    if direct:
        return ("direct", "no application sid, so the number's own url is read: "
                + ", ".join(direct))
    return ("idle", "no voice or sms handler and no application sid")


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 and cache it by SID."""
    sids = set()
    for n in numbers:
        for _channel, _url_field, app_field in CHANNELS:
            sid = str(n.get(app_field) or "").strip()
            if sid:
                sids.add(sid)
    return {sid: get(session, "%s/Accounts/%s/Applications/%s.json"
                     % (BASE, account, sid))
            for sid in sorted(sids)}


def sharing(numbers, app_sid):
    """Every number attached to one app. Pure, and the reason it exists is that
    editing an app moves all of them at once."""
    out = []
    for n in numbers:
        for _channel, _url_field, app_field in CHANNELS:
            if str(n.get(app_field) or "").strip() == app_sid:
                out.append(n.get("phone_number") or n.get("sid"))
                break
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-numbers", type=int, default=1000,
                    help="stop paging after this many numbers")
    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 = "%-14s %s  %s" % (state, n.get("phone_number", "?"), detail)
        if state in ("direct", "app-routed", "idle"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        for _channel, _url_field, app_field in CHANNELS:
            sid = str(n.get(app_field) or "").strip()
            if not sid:
                continue
            peers = sharing(numbers, sid)
            log.warning("  app %s also fronts %d number(s): %s",
                        sid, len(peers), ", ".join(str(p) for p in peers[:5]))
        log.warning("  repair: either update the app, POST %s/Accounts/%s/"
                    "Applications/{AppSid}.json VoiceUrl=https://.../voice, which "
                    "moves every number above; or detach it, POST %s/Accounts/%s/"
                    "IncomingPhoneNumbers/%s.json VoiceApplicationSid= (empty), "
                    "so the number's own voice_url is read again.",
                    BASE, account, BASE, account, n.get("sid"))

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


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

/**
 * Report Twilio numbers whose webhook URLs are shadowed by an Application SID.
 *
 * 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`;

// [channel, url field, application sid field]. The Application resource names
// its URLs identically, which is what lets one comparison serve both.
const CHANNELS = [
  ['voice', 'voice_url', 'voice_application_sid'],
  ['sms', 'sms_url', 'sms_application_sid'],
];

/**
 * Classify one IncomingPhoneNumber against the apps it references. Pure, so the
 * precedence rule is testable without a network. `apps` maps an Application SID
 * to that Application: when a channel carries one, the Application is the
 * effective handler and the number's own url is never requested.
 * Returns [state, detail].
 */
export function verdict(number, apps = {}) {
  const unresolved = [];
  const dead = [];
  const shadowed = [];
  const routed = [];
  const direct = [];

  for (const [channel, urlField, appField] of CHANNELS) {
    const appSid = String(number[appField] ?? '').trim();
    const own = String(number[urlField] ?? '').trim();

    if (!appSid) {
      if (own) direct.push(`${channel} serves ${own}`);
      continue;
    }

    const app = apps[appSid];
    if (app === undefined) { unresolved.push(`${channel} (${appSid})`); continue; }

    const live = String(app[urlField] ?? '').trim();
    if (!live) { dead.push(`${channel}: app ${appSid} has no ${urlField}`); continue; }
    if (own && own !== live) {
      shadowed.push(`${channel}: ${own} on the number is ignored, app ${appSid} serves ${live}`);
      continue;
    }
    routed.push(`${channel} via app ${appSid}`);
  }

  if (unresolved.length) {
    return ['unresolved',
      `an application sid is set but that application was not read: ${unresolved.join(', ')}`];
  }
  if (dead.length) {
    return ['routes-nowhere',
      `${dead.join('; ')}. The number's own url cannot rescue this: the app wins ` +
      'while it is attached.'];
  }
  if (shadowed.length) {
    return ['shadowed', `${shadowed.join('; ')}. Editing the number changes nothing.`];
  }
  if (routed.length) return ['app-routed', `handled by its application: ${routed.join(', ')}`];
  if (direct.length) {
    return ['direct', `no application sid, so the number's own url is read: ${direct.join(', ')}`];
  }
  return ['idle', 'no voice or sms handler and no application sid'];
}

/** Every number attached to one app. Pure: editing an app moves all of them. */
export function sharing(numbers, appSid) {
  const out = [];
  for (const n of numbers) {
    for (const [, , appField] of CHANNELS) {
      if (String(n[appField] ?? '').trim() === appSid) {
        out.push(n.phone_number ?? n.sid);
        break;
      }
    }
  }
  return out;
}

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(14)} ${n.phone_number ?? '?'}  ${detail}`;
    if (state === 'direct' || state === 'app-routed' || state === 'idle') {
      console.log(line);
      continue;
    }
    bad += 1;
    console.warn(line);
    for (const [, , appField] of CHANNELS) {
      const sid = String(n[appField] ?? '').trim();
      if (!sid) continue;
      const peers = sharing(numbers, sid);
      console.warn(`  app ${sid} also fronts ${peers.length} number(s): ` +
                   `${peers.slice(0, 5).join(', ')}`);
    }
    console.warn(`  repair: either update the app, POST ${BASE}/Accounts/${account}` +
                 '/Applications/{AppSid}.json VoiceUrl=https://.../voice, which moves ' +
                 `every number above; or detach it, POST ${BASE}/Accounts/${account}` +
                 `/IncomingPhoneNumbers/${n.sid}.json VoiceApplicationSid= (empty), ` +
                 "so the number's own voice_url is read again.");
  }

  console.log(`${numbers.length} number(s), ${bad} with a shadowed 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_number_app_precedence_audit import sharing, verdict

APP = "AP11111111111111111111111111111111"
OTHER = "AP22222222222222222222222222222222"


def test_a_different_url_on_the_number_is_shadowed():
    # The whole note: this number looks configured and the field is inert.
    state, detail = verdict(
        {"voice_application_sid": APP, "voice_url": "https://new.example.com/voice"},
        {APP: {"voice_url": "https://retired.example.com/voice"}})
    assert state == "shadowed"
    assert "retired.example.com" in detail
    assert "Editing the number changes nothing" in detail


def test_the_same_url_on_both_is_not_a_finding():
    state, _ = verdict(
        {"voice_application_sid": APP, "voice_url": "https://app.example.com/voice"},
        {APP: {"voice_url": "https://app.example.com/voice"}})
    assert state == "app-routed"


def test_an_application_with_no_url_routes_nowhere():
    state, detail = verdict(
        {"voice_application_sid": APP, "voice_url": "https://app.example.com/voice"},
        {APP: {"voice_url": ""}})
    assert state == "routes-nowhere"
    assert "has no voice_url" in detail


def test_sms_precedence_is_checked_independently():
    state, detail = verdict(
        {"voice_url": "https://app.example.com/voice",
         "sms_application_sid": APP, "sms_url": "https://new.example.com/sms"},
        {APP: {"sms_url": "https://retired.example.com/sms"}})
    assert state == "shadowed"
    assert "sms:" in detail


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


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


def test_a_number_with_nothing_configured_is_idle():
    assert verdict({"voice_url": "", "sms_url": None})[0] == "idle"


def test_sharing_lists_every_number_on_one_app_once():
    numbers = [
        {"phone_number": "+15550001111", "voice_application_sid": APP,
         "sms_application_sid": APP},
        {"phone_number": "+15550002222", "sms_application_sid": APP},
        {"phone_number": "+15550003333", "voice_application_sid": OTHER},
    ]
    assert sharing(numbers, APP) == ["+15550001111", "+15550002222"]
    assert sharing(numbers, OTHER) == ["+15550003333"]

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { sharing, verdict } from './twilio-number-app-precedence-audit.mjs';

const APP = 'AP11111111111111111111111111111111';
const OTHER = 'AP22222222222222222222222222222222';

test('a different url on the number is shadowed', () => {
  const [state, detail] = verdict(
    { voice_application_sid: APP, voice_url: 'https://new.example.com/voice' },
    { [APP]: { voice_url: 'https://retired.example.com/voice' } });
  assert.equal(state, 'shadowed');
  assert.match(detail, /retired\.example\.com/);
  assert.match(detail, /Editing the number changes nothing/);
});

test('the same url on both is not a finding', () => {
  const [state] = verdict(
    { voice_application_sid: APP, voice_url: 'https://app.example.com/voice' },
    { [APP]: { voice_url: 'https://app.example.com/voice' } });
  assert.equal(state, 'app-routed');
});

test('an application with no url routes nowhere', () => {
  const [state, detail] = verdict(
    { voice_application_sid: APP, voice_url: 'https://app.example.com/voice' },
    { [APP]: { voice_url: '' } });
  assert.equal(state, 'routes-nowhere');
  assert.match(detail, /has no voice_url/);
});

test('sms precedence is checked independently', () => {
  const [state, detail] = verdict(
    { voice_url: 'https://app.example.com/voice',
      sms_application_sid: APP, sms_url: 'https://new.example.com/sms' },
    { [APP]: { sms_url: 'https://retired.example.com/sms' } });
  assert.equal(state, 'shadowed');
  assert.match(detail, /sms:/);
});

test('no application sid means the number is read', () => {
  const [state, detail] = verdict({ voice_url: 'https://app.example.com/voice' });
  assert.equal(state, 'direct');
  assert.match(detail, /app\.example\.com/);
});

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

test('a number with nothing configured is idle', () => {
  assert.equal(verdict({ voice_url: '', sms_url: null })[0], 'idle');
});

test('sharing lists every number on one app once', () => {
  const numbers = [
    { phone_number: '+15550001111', voice_application_sid: APP, sms_application_sid: APP },
    { phone_number: '+15550002222', sms_application_sid: APP },
    { phone_number: '+15550003333', voice_application_sid: OTHER },
  ];
  assert.deepEqual(sharing(numbers, APP), ['+15550001111', '+15550002222']);
  assert.deepEqual(sharing(numbers, OTHER), ['+15550003333']);
});