numbers still pinned to the 2008-08-01 API version

"""Report Twilio phone numbers pinned to an older API version.

The pin is per number and set at purchase time. It does not expire and nothing
migrates it, so a number bought in 2014 is still served the 2008 schema today:
webhooks with fewer parameters, and resource fields the current documentation
promises simply absent.

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

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

CURRENT = "2010-04-01"
LEGACY = "2008-08-01"

ROUTING_FIELDS = ("voice_url", "sms_url", "voice_fallback_url", "sms_fallback_url",
                  "status_callback", "voice_application_sid", "sms_application_sid")


def is_routed(number):
    """True when something on this number would actually be fetched.

    The version pin only reaches your application through a webhook, so a pinned
    number with no handler on it is a different sentence from a pinned number
    that is answering calls today. Both are findings; only one is current.
    """
    return any(str(number.get(f) or "").strip() for f in ROUTING_FIELDS)


def verdict(number):
    """Classify one IncomingPhoneNumber by the API version it is pinned to.

    Pure, so the rules can be tested without a network. Returns (state, detail).
    """
    version = str(number.get("api_version") or "").strip()

    if not version:
        return ("unread",
                "no api_version on this resource: report it rather than assuming "
                "it is current, because an unknown quietly counted as fine is how "
                "the one number that matters gets skipped.")

    if version == CURRENT:
        return ("current", "on %s, the version the documentation describes." % CURRENT)

    if version == LEGACY:
        if is_routed(number):
            return ("legacy-live",
                    "pinned to %s and wired to a handler: every webhook Twilio "
                    "sends for this number is built from the %s schema, so "
                    "parameters the docs promise arrive absent rather than wrong."
                    % (LEGACY, LEGACY))
        return ("legacy-idle",
                "pinned to %s with no handler on it: nothing is receiving the old "
                "schema today, and something will on the day this number is used."
                % LEGACY)

    return ("unread",
            "api_version is %s, which is neither %s nor %s: read it before "
            "assuming anything about what the webhooks carry."
            % (version, CURRENT, LEGACY))


def account_verdict(account):
    """Classify the account's default API version. Pure.

    Separate from the per-number check because this field decides what the next
    number bought on this account arrives pinned to. Repairing the numbers and
    leaving this one is a treadmill: the audit passes this quarter and fails the
    next, for a reason nobody wrote down.

    Returns (state, detail).
    """
    version = str(account.get("api_version") or "").strip()
    if not version:
        return ("unread",
                "no api_version on the account resource: the default that new "
                "numbers inherit could not be read.")
    if version == CURRENT:
        return ("current", "account default is %s." % CURRENT)
    return ("legacy-default",
            "account default is %s: every number bought from here on arrives "
            "pinned to it, so repairing the numbers alone fixes nothing that "
            "stays fixed." % version)


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

    state, detail = account_verdict(get(session, "%s/Accounts/%s.json" % (BASE, account)))
    bad = 0
    if state == "current":
        log.info("%-14s %s", state, detail)
    else:
        bad += 1
        log.warning("%-14s %s", state, detail)
        log.warning("  repair: Console > Account > API version, set it to %s", CURRENT)

    numbers = list_numbers(session, account, args.max_numbers)
    pinned = 0
    for n in numbers:
        state, detail = verdict(n)
        line = "%-14s %s  %s" % (state, n.get("phone_number", "?"), detail)
        if state == "current":
            log.info(line)
            continue
        pinned += 1
        log.warning(line)
        log.warning("  repair: POST %s/Accounts/%s/IncomingPhoneNumbers/%s.json "
                    "ApiVersion=%s", BASE, account, n.get("sid"), CURRENT)

    log.info("%d number(s), %d pinned to an older API version", len(numbers), pinned)
    return 1 if (pinned or bad) else 0


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

/**
 * Report Twilio phone numbers pinned to an older API version.
 *
 * The pin is per number and set at purchase time. It does not expire and nothing
 * migrates it, so a number bought in 2014 is still served the 2008 schema today.
 *
 * 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 CURRENT = '2010-04-01';
const LEGACY = '2008-08-01';

const ROUTING_FIELDS = ['voice_url', 'sms_url', 'voice_fallback_url',
                        'sms_fallback_url', 'status_callback',
                        'voice_application_sid', 'sms_application_sid'];

/**
 * True when something on this number would actually be fetched. The version pin
 * only reaches your application through a webhook.
 */
export function isRouted(number) {
  return ROUTING_FIELDS.some((f) => String(number[f] ?? '').trim());
}

/**
 * Classify one IncomingPhoneNumber by the API version it is pinned to. Pure, so
 * the rules can be tested without a network. Returns [state, detail].
 */
export function verdict(number) {
  const version = String(number.api_version ?? '').trim();

  if (!version) {
    return ['unread',
      'no api_version on this resource: report it rather than assuming it is ' +
      'current, because an unknown quietly counted as fine is how the one ' +
      'number that matters gets skipped.'];
  }

  if (version === CURRENT) {
    return ['current', `on ${CURRENT}, the version the documentation describes.`];
  }

  if (version === LEGACY) {
    if (isRouted(number)) {
      return ['legacy-live',
        `pinned to ${LEGACY} and wired to a handler: every webhook Twilio sends ` +
        `for this number is built from the ${LEGACY} schema, so parameters the ` +
        'docs promise arrive absent rather than wrong.'];
    }
    return ['legacy-idle',
      `pinned to ${LEGACY} with no handler on it: nothing is receiving the old ` +
      'schema today, and something will on the day this number is used.'];
  }

  return ['unread',
    `api_version is ${version}, which is neither ${CURRENT} nor ${LEGACY}: read ` +
    'it before assuming anything about what the webhooks carry.'];
}

/**
 * Classify the account's default API version. Pure. This field decides what the
 * next number bought on this account arrives pinned to, so repairing the numbers
 * and leaving it is a treadmill. Returns [state, detail].
 */
export function accountVerdict(account) {
  const version = String(account.api_version ?? '').trim();
  if (!version) {
    return ['unread',
      'no api_version on the account resource: the default that new numbers ' +
      'inherit could not be read.'];
  }
  if (version === CURRENT) return ['current', `account default is ${CURRENT}.`];
  return ['legacy-default',
    `account default is ${version}: every number bought from here on arrives ` +
    'pinned to it, so repairing the numbers alone fixes nothing that stays fixed.'];
}

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

  let bad = 0;
  const [aState, aDetail] = accountVerdict(
    await get(auth, `${BASE}/Accounts/${account}.json`));
  if (aState === 'current') {
    console.log(`${aState.padEnd(14)} ${aDetail}`);
  } else {
    bad += 1;
    console.warn(`${aState.padEnd(14)} ${aDetail}`);
    console.warn(`  repair: Console > Account > API version, set it to ${CURRENT}`);
  }

  const numbers = await listNumbers(auth, account);
  let pinned = 0;
  for (const n of numbers) {
    const [state, detail] = verdict(n);
    const line = `${state.padEnd(14)} ${n.phone_number ?? '?'}  ${detail}`;
    if (state === 'current') { console.log(line); continue; }
    pinned += 1;
    console.warn(line);
    console.warn(`  repair: POST ${BASE}/Accounts/${account}/IncomingPhoneNumbers/` +
                 `${n.sid}.json ApiVersion=${CURRENT}`);
  }

  console.log(`${numbers.length} number(s), ${pinned} pinned to an older API version`);
  process.exitCode = (pinned || 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_api_version_audit import account_verdict, is_routed, verdict


def make(**kw):
    number = {"sid": "PN01", "phone_number": "+15005550006",
              "api_version": "2010-04-01", "voice_url": "https://app.example.com/voice"}
    number.update(kw)
    return number


def test_a_number_on_the_current_version_is_current():
    state, detail = verdict(make())
    assert state == "current"
    assert "2010-04-01" in detail


def test_a_2008_pin_with_a_live_handler_is_serving_the_old_schema_now():
    state, detail = verdict(make(api_version="2008-08-01"))
    assert state == "legacy-live"
    assert "absent" in detail


def test_a_2008_pin_with_no_handler_is_a_separate_and_quieter_finding():
    state, detail = verdict(make(api_version="2008-08-01", voice_url=""))
    assert state == "legacy-idle"
    assert "day this number is used" in detail


def test_an_application_sid_alone_still_counts_as_routed():
    assert is_routed({"voice_application_sid": "AP0123456789"}) is True
    assert is_routed({"voice_url": "", "sms_url": None}) is False


def test_a_missing_api_version_is_reported_rather_than_assumed_current():
    state, detail = verdict(make(api_version=None))
    assert state == "unread"
    assert "assuming" in detail


def test_an_unexpected_version_is_never_folded_into_either_bucket():
    state, detail = verdict(make(api_version="2015-01-01"))
    assert state == "unread"
    assert "2015-01-01" in detail


def test_the_account_default_is_its_own_finding_with_its_own_repair():
    state, detail = account_verdict({"api_version": "2008-08-01"})
    assert state == "legacy-default"
    assert "bought from here on" in detail


def test_a_current_account_default_means_new_numbers_arrive_correct():
    assert account_verdict({"api_version": "2010-04-01"})[0] == "current"
    assert account_verdict({})[0] == "unread"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { accountVerdict, isRouted, verdict } from './twilio-api-version-audit.mjs';

const make = (over = {}) => ({
  sid: 'PN01',
  phone_number: '+15005550006',
  api_version: '2010-04-01',
  voice_url: 'https://app.example.com/voice',
  ...over,
});

test('a number on the current version is current', () => {
  const [state, detail] = verdict(make());
  assert.equal(state, 'current');
  assert.match(detail, /2010-04-01/);
});

test('a 2008 pin with a live handler is serving the old schema now', () => {
  const [state, detail] = verdict(make({ api_version: '2008-08-01' }));
  assert.equal(state, 'legacy-live');
  assert.match(detail, /absent/);
});

test('a 2008 pin with no handler is a separate and quieter finding', () => {
  const [state, detail] = verdict(make({ api_version: '2008-08-01', voice_url: '' }));
  assert.equal(state, 'legacy-idle');
  assert.match(detail, /day this number is used/);
});

test('an application sid alone still counts as routed', () => {
  assert.equal(isRouted({ voice_application_sid: 'AP0123456789' }), true);
  assert.equal(isRouted({ voice_url: '', sms_url: null }), false);
});

test('a missing api_version is reported rather than assumed current', () => {
  const [state, detail] = verdict(make({ api_version: null }));
  assert.equal(state, 'unread');
  assert.match(detail, /assuming/);
});

test('an unexpected version is never folded into either bucket', () => {
  const [state, detail] = verdict(make({ api_version: '2015-01-01' }));
  assert.equal(state, 'unread');
  assert.match(detail, /2015-01-01/);
});

test('the account default is its own finding with its own repair', () => {
  const [state, detail] = accountVerdict({ api_version: '2008-08-01' });
  assert.equal(state, 'legacy-default');
  assert.match(detail, /bought from here on/);
});

test('a current account default means new numbers arrive correct', () => {
  assert.equal(accountVerdict({ api_version: '2010-04-01' })[0], 'current');
  assert.equal(accountVerdict({})[0], 'unread');
});