an unverified toll-free number is blocked, not throttled

"""Report toll-free numbers that cannot send US or CA SMS for want of verification.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The submission 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_tollfree_verification_audit")

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

# Since 31 January 2024 these are blocked, not throttled. They belong in the
# same report as a number with no verification record at all.
BLOCKED_REVIEW = ("PENDING_REVIEW", "IN_REVIEW")


def pick_verification(records):
    """Choose the record that governs a number. Pure.

    A number can carry more than one: an old rejection and a newer approval, for
    instance. A plain set-membership check reports whichever the API returned
    first, so prefer TWILIO_APPROVED and otherwise take the most recently
    updated.
    """
    if not records:
        return None
    approved = [r for r in records
                if str(r.get("status") or "").upper() == "TWILIO_APPROVED"]
    pool = approved or list(records)
    return max(pool, key=lambda r: str(r.get("date_updated")
                                       or r.get("date_created") or ""))


def rejection_lines(verification):
    """Why a verification was rejected, from the structured fields first. Pure.

    rejection_reasons[] is the list nobody reads; error_code and the
    rejection_reason prose are the fallbacks when it is absent.
    """
    lines = []
    for reason in verification.get("rejection_reasons") or []:
        code = reason.get("code") or reason.get("error_code") or "no code"
        lines.append("%s: %s" % (code, reason.get("description")
                                 or "no description"))
    if lines:
        return lines
    code = verification.get("error_code")
    prose = str(verification.get("rejection_reason") or "").strip()
    if code or prose:
        lines.append("%s: %s" % (code or "no code", prose or "no description"))
    return lines


def verdict(number, verification):
    """Decide whether one toll-free number can send US or CA SMS. Pure, so the
    blocked states can be tested without a network.

    Returns (state, detail).
    """
    if not (number.get("capabilities") or {}).get("sms"):
        return ("voice-only",
                "toll-free number with no SMS capability: nothing to verify.")

    if not verification:
        return ("unverified",
                "no toll-free verification record at all. Every US or CA SMS "
                "from this number fails 30032, and the attempts are billed.")

    status = str(verification.get("status") or "").upper()

    if status == "TWILIO_APPROVED":
        return ("verified", "verification %s is TWILIO_APPROVED"
                % (verification.get("sid") or "?"))

    if status in BLOCKED_REVIEW:
        return ("blocked-in-review",
                "verification is %s. Filing is not passing: since 31 January "
                "2024 traffic in a review state is blocked outright rather than "
                "throttled." % status)

    if status == "TWILIO_REJECTED":
        reasons = "; ".join(rejection_lines(verification)) or "no reason on the record"
        if verification.get("edit_allowed"):
            return ("rejected-editable",
                    "rejected (%s). edit_allowed is true until %s, so the named "
                    "fields can still be corrected in place."
                    % (reasons, verification.get("edit_expiration") or "an "
                       "unstated date"))
        return ("rejected-final",
                "rejected (%s) and edit_allowed is false: a fresh submission is "
                "the only path, at the back of the review queue." % reasons)

    return ("unknown-status",
            "verification status is %s, which this script does not recognise."
            % (status or "unset"))


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_tollfree(session, account, limit=1000):
    """Page the toll-free numbers. next_page_uri is a path, not an absolute URL."""
    url = "%s/Accounts/%s/IncomingPhoneNumbers/TollFree.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 list_verifications(session, limit=1000):
    """Page the toll-free verifications. meta.next_page_url is absolute."""
    url = MSG + "/Tollfree/Verifications"
    out = []
    while url and len(out) < limit:
        page = get(session, url, PageSize=50)
        out.extend(page.get("verifications", []))
        url = (page.get("meta") or {}).get("next_page_url")
    return out[:limit]


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_tollfree(session, account, args.max_numbers)
    if not numbers:
        log.info("no toll-free numbers on this account")
        return 0

    by_sid = {}
    for record in list_verifications(session):
        by_sid.setdefault(record.get("tollfree_phone_number_sid"), []).append(record)

    bad = 0
    for n in numbers:
        verification = pick_verification(by_sid.get(n.get("sid")) or [])
        state, detail = verdict(n, verification)
        line = "%-18s %s  %s" % (state, n.get("phone_number", "?"), detail)
        if state in ("verified", "voice-only"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if state == "unverified":
            log.warning("  repair: POST %s/Tollfree/Verifications with BusinessName, "
                        "BusinessWebsite, NotificationEmail, UseCaseCategories, "
                        "UseCaseSummary, ProductionMessageSample, OptInType, "
                        "OptInImageUrls, MessageVolume and "
                        "TollfreePhoneNumberSid=%s", MSG, n.get("sid", "PN..."))
        elif state == "rejected-editable":
            log.warning("  repair: POST %s/Tollfree/Verifications/%s correcting the "
                        "named fields before edit_expiration", MSG,
                        verification.get("sid", "HH..."))
        elif state == "blocked-in-review":
            log.warning("  repair: none by API. Wait for TWILIO_APPROVED and do not "
                        "route production traffic through this number meanwhile")

    log.info("%d toll-free number(s), %d blocked from US and CA SMS",
             len(numbers), bad)
    return 1 if bad else 0


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

/**
 * Report toll-free numbers that cannot send US or CA SMS for want of verification.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The submission is printed, never
 * performed.
 */
const HOST = 'https://api.twilio.com';
const BASE = `${HOST}/2010-04-01`;
const MSG = 'https://messaging.twilio.com/v1';

// Since 31 January 2024 these are blocked, not throttled.
const BLOCKED_REVIEW = ['PENDING_REVIEW', 'IN_REVIEW'];

/**
 * Choose the record that governs a number. Pure. A number can carry more than
 * one, so prefer TWILIO_APPROVED and otherwise take the most recently updated.
 */
export function pickVerification(records) {
  if (!records || records.length === 0) return null;
  const approved = records.filter(
    (r) => String(r.status ?? '').toUpperCase() === 'TWILIO_APPROVED');
  const pool = approved.length ? approved : records;
  const stamp = (r) => String(r.date_updated ?? r.date_created ?? '');
  return pool.reduce((best, r) => (stamp(r) > stamp(best) ? r : best), pool[0]);
}

/** Why a verification was rejected, from the structured fields first. Pure. */
export function rejectionLines(verification) {
  const lines = [];
  for (const reason of verification.rejection_reasons ?? []) {
    const code = reason.code ?? reason.error_code ?? 'no code';
    lines.push(`${code}: ${reason.description ?? 'no description'}`);
  }
  if (lines.length) return lines;
  const code = verification.error_code;
  const prose = String(verification.rejection_reason ?? '').trim();
  if (code || prose) lines.push(`${code ?? 'no code'}: ${prose || 'no description'}`);
  return lines;
}

/**
 * Decide whether one toll-free number can send US or CA SMS. Pure, so the
 * blocked states can be tested without a network. Returns [state, detail].
 */
export function verdict(number, verification) {
  if (!(number.capabilities ?? {}).sms) {
    return ['voice-only', 'toll-free number with no SMS capability: nothing to verify.'];
  }

  if (!verification) {
    return ['unverified',
      'no toll-free verification record at all. Every US or CA SMS from this ' +
      'number fails 30032, and the attempts are billed.'];
  }

  const status = String(verification.status ?? '').toUpperCase();

  if (status === 'TWILIO_APPROVED') {
    return ['verified', `verification ${verification.sid ?? '?'} is TWILIO_APPROVED`];
  }

  if (BLOCKED_REVIEW.includes(status)) {
    return ['blocked-in-review',
      `verification is ${status}. Filing is not passing: since 31 January 2024 ` +
      'traffic in a review state is blocked outright rather than throttled.'];
  }

  if (status === 'TWILIO_REJECTED') {
    const reasons = rejectionLines(verification).join('; ') || 'no reason on the record';
    if (verification.edit_allowed) {
      return ['rejected-editable',
        `rejected (${reasons}). edit_allowed is true until ` +
        `${verification.edit_expiration ?? 'an unstated date'}, so the named ` +
        'fields can still be corrected in place.'];
    }
    return ['rejected-final',
      `rejected (${reasons}) and edit_allowed is false: a fresh submission is ` +
      'the only path, at the back of the review queue.'];
  }

  return ['unknown-status',
    `verification status is ${status || 'unset'}, which this script does not recognise.`];
}

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 listTollfree(auth, account, limit = 1000) {
  let url = `${BASE}/Accounts/${account}/IncomingPhoneNumbers/TollFree.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);
}

export async function listVerifications(auth, limit = 1000) {
  const out = [];
  let next = `${MSG}/Tollfree/Verifications`;
  while (next && out.length < limit) {
    const page = await get(auth, next, { PageSize: 50 });
    out.push(...(page.verifications ?? []));
    next = page.meta?.next_page_url ?? null;
  }
  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 numbers = await listTollfree(auth, account);
  if (numbers.length === 0) {
    console.log('no toll-free numbers on this account');
    return;
  }

  const bySid = new Map();
  for (const record of await listVerifications(auth)) {
    const sid = record.tollfree_phone_number_sid;
    if (!bySid.has(sid)) bySid.set(sid, []);
    bySid.get(sid).push(record);
  }

  let bad = 0;
  for (const n of numbers) {
    const verification = pickVerification(bySid.get(n.sid) ?? []);
    const [state, detail] = verdict(n, verification);
    const line = `${state.padEnd(18)} ${n.phone_number ?? '?'}  ${detail}`;
    if (state === 'verified' || state === 'voice-only') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    if (state === 'unverified') {
      console.warn(`  repair: POST ${MSG}/Tollfree/Verifications with BusinessName, ` +
                   'BusinessWebsite, NotificationEmail, UseCaseCategories, ' +
                   'UseCaseSummary, ProductionMessageSample, OptInType, ' +
                   `OptInImageUrls, MessageVolume and TollfreePhoneNumberSid=${n.sid}`);
    } else if (state === 'rejected-editable') {
      console.warn(`  repair: POST ${MSG}/Tollfree/Verifications/` +
                   `${verification.sid ?? 'HH...'} correcting the named fields ` +
                   'before edit_expiration');
    } else if (state === 'blocked-in-review') {
      console.warn('  repair: none by API. Wait for TWILIO_APPROVED and do not route ' +
                   'production traffic through this number meanwhile');
    }
  }

  console.log(`${numbers.length} toll-free number(s), ${bad} blocked from US and CA SMS`);
  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_tollfree_verification_audit import pick_verification, verdict

SMS = {"sid": "PN0123456789", "phone_number": "+18885551234",
       "capabilities": {"sms": True, "voice": True}}


def test_no_verification_record_is_the_headline_finding():
    state, detail = verdict(SMS, None)
    assert state == "unverified"
    assert "30032" in detail


def test_pending_review_is_blocked_not_progress():
    # The point of the note: filing is not passing.
    state, detail = verdict(SMS, {"status": "PENDING_REVIEW"})
    assert state == "blocked-in-review"
    assert "blocked outright" in detail


def test_approved_is_the_only_state_that_can_send():
    state, detail = verdict(SMS, {"status": "TWILIO_APPROVED", "sid": "HH0123456789"})
    assert state == "verified"
    assert "HH0123456789" in detail


def test_rejection_reasons_are_read_from_the_array():
    state, detail = verdict(SMS, {
        "status": "TWILIO_REJECTED", "edit_allowed": True,
        "edit_expiration": "2026-09-05T00:00:00Z",
        "rejection_reasons": [{"code": 30469,
                               "description": "Illegal substances or articles"}]})
    assert state == "rejected-editable"
    assert "30469" in detail
    assert "2026-09-05" in detail


def test_rejection_falls_back_to_the_prose_field():
    state, detail = verdict(SMS, {"status": "TWILIO_REJECTED", "edit_allowed": False,
                                  "rejection_reason": "opt-in evidence missing"})
    assert state == "rejected-final"
    assert "opt-in evidence missing" in detail


def test_a_voice_only_toll_free_number_is_not_a_finding():
    state, _ = verdict({"capabilities": {"sms": False, "voice": True}}, None)
    assert state == "voice-only"


def test_an_approved_record_wins_over_a_newer_rejection():
    records = [{"status": "TWILIO_APPROVED", "date_updated": "2026-01-01T00:00:00Z"},
               {"status": "TWILIO_REJECTED", "date_updated": "2026-06-01T00:00:00Z"}]
    assert pick_verification(records)["status"] == "TWILIO_APPROVED"


def test_without_an_approval_the_newest_record_governs():
    records = [{"status": "TWILIO_REJECTED", "date_updated": "2026-01-01T00:00:00Z"},
               {"status": "PENDING_REVIEW", "date_updated": "2026-06-01T00:00:00Z"}]
    assert pick_verification(records)["status"] == "PENDING_REVIEW"
    assert pick_verification([]) is None

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { pickVerification, verdict } from './twilio-tollfree-verification-audit.mjs';

const SMS = { sid: 'PN0123456789', phone_number: '+18885551234',
              capabilities: { sms: true, voice: true } };

test('no verification record is the headline finding', () => {
  const [state, detail] = verdict(SMS, null);
  assert.equal(state, 'unverified');
  assert.match(detail, /30032/);
});

test('pending review is blocked, not progress', () => {
  const [state, detail] = verdict(SMS, { status: 'PENDING_REVIEW' });
  assert.equal(state, 'blocked-in-review');
  assert.match(detail, /blocked outright/);
});

test('approved is the only state that can send', () => {
  const [state, detail] = verdict(SMS, { status: 'TWILIO_APPROVED',
                                         sid: 'HH0123456789' });
  assert.equal(state, 'verified');
  assert.match(detail, /HH0123456789/);
});

test('rejection reasons are read from the array', () => {
  const [state, detail] = verdict(SMS, {
    status: 'TWILIO_REJECTED', edit_allowed: true,
    edit_expiration: '2026-09-05T00:00:00Z',
    rejection_reasons: [{ code: 30469, description: 'Illegal substances or articles' }],
  });
  assert.equal(state, 'rejected-editable');
  assert.match(detail, /30469/);
  assert.match(detail, /2026-09-05/);
});

test('rejection falls back to the prose field', () => {
  const [state, detail] = verdict(SMS, { status: 'TWILIO_REJECTED',
                                         edit_allowed: false,
                                         rejection_reason: 'opt-in evidence missing' });
  assert.equal(state, 'rejected-final');
  assert.match(detail, /opt-in evidence missing/);
});

test('a voice only toll free number is not a finding', () => {
  assert.equal(verdict({ capabilities: { sms: false, voice: true } }, null)[0],
               'voice-only');
});

test('an approved record wins over a newer rejection', () => {
  const records = [{ status: 'TWILIO_APPROVED', date_updated: '2026-01-01T00:00:00Z' },
                   { status: 'TWILIO_REJECTED', date_updated: '2026-06-01T00:00:00Z' }];
  assert.equal(pickVerification(records).status, 'TWILIO_APPROVED');
});

test('without an approval the newest record governs', () => {
  const records = [{ status: 'TWILIO_REJECTED', date_updated: '2026-01-01T00:00:00Z' },
                   { status: 'PENDING_REVIEW', date_updated: '2026-06-01T00:00:00Z' }];
  assert.equal(pickVerification(records).status, 'PENDING_REVIEW');
  assert.equal(pickVerification([]), null);
});