error 21703: the pool has senders but none reach the To

"""Report Messaging Services whose sender pool cannot reach a destination.

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

MESSAGING = "https://messaging.twilio.com/v1"
LOOKUPS = "https://lookups.twilio.com/v2"

# Alphanumeric sender IDs are not candidates for these destinations, so a pool
# holding nothing else is uncovered however full it looks.
ALPHA_EXCLUDED = ("US", "CA")


def has_capability(entry, name):
    """Case-insensitive capability test that survives both spellings.

    The Messaging Service pool returns a list like ["SMS", "MMS", "voice"]. The
    account API returns an object with lowercase keys for the same facts.
    Comparing raw strings across the two is how an MMS-capable pool gets
    reported as having no MMS sender.
    """
    caps = entry.get("capabilities") or []
    if isinstance(caps, dict):
        return bool(caps.get(name.lower()))
    return name.lower() in [str(c).lower() for c in caps]


def coverage(pool, destination):
    """Decide whether one sender pool can reach one destination.

    `pool` is {"phone_numbers": [...], "short_codes": [...], "alpha_senders": [...]}
    as returned by the three subresources. `destination` is
    {"country_code": "US", "needs_mms": bool}.

    Pure, so the matching rule can be tested without a network.
    Returns (state, detail).
    """
    country = str(destination.get("country_code") or "").upper()
    needs_mms = bool(destination.get("needs_mms"))
    numbers = pool.get("phone_numbers") or []
    codes = pool.get("short_codes") or []
    alphas = pool.get("alpha_senders") or []

    if not (numbers or codes or alphas):
        return ("no-senders",
                "the pool holds no senders at all, which is 21704 on every send "
                "rather than 21703 on this destination.")
    if not country:
        return ("unresolved",
                "the destination country was not resolved, so coverage cannot "
                "be decided. Read country_code from Lookup v2 first.")

    local = [n for n in numbers
             if str(n.get("country_code") or "").upper() == country]
    local_codes = [c for c in codes
                   if str(c.get("country_code") or "").upper() == country]

    if not (local or local_codes):
        if country in ALPHA_EXCLUDED:
            return ("unreachable",
                    "no %s number or short code in the pool. The %d alphanumeric "
                    "sender(s) do not count: they cannot deliver to %s."
                    % (country, len(alphas), country))
        if alphas:
            return ("alpha-only",
                    "no %s number in the pool, only %d alphanumeric sender(s). "
                    "They are one way and are not accepted everywhere, so this "
                    "is deliverable in some countries and 21703 in others."
                    % (country, len(alphas)))
        return ("no-local-sender",
                "no %s sender in the pool. Selection may still pick a foreign "
                "long code, and this is the shape that returns 21703 when it "
                "does not." % country)

    if needs_mms and not any(has_capability(n, "MMS") for n in local):
        return ("no-mms",
                "%d %s sender(s) in the pool and not one of them lists MMS, so "
                "any message carrying MediaUrl is 21703 while the text only "
                "version of it sends." % (len(local), country))

    kinds = []
    if local:
        kinds.append("%d number(s)" % len(local))
    if local_codes:
        kinds.append("%d short code(s)" % len(local_codes))
    return ("covered", "%s in %s%s"
            % (", ".join(kinds), country, ", MMS capable" if needs_mms else ""))


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_services(session, limit):
    url = "%s/Services" % MESSAGING
    params = {"PageSize": 100}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("services", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def read_pool(session, service_sid):
    """All three sender lists. Reading one and concluding from it is how a
    short code only service gets reported as broken."""
    pool = {}
    for path, key in (("PhoneNumbers", "phone_numbers"),
                      ("ShortCodes", "short_codes"),
                      ("AlphaSenders", "alpha_senders")):
        page = get(session, "%s/Services/%s/%s" % (MESSAGING, service_sid, path),
                   PageSize=100)
        pool[key] = page.get(key, [])
    return pool


def resolve(session, e164):
    """Destination country from Lookup v2. Prefix arithmetic breaks on +1,
    which covers the US, Canada and several Caribbean countries."""
    page = get(session, "%s/PhoneNumbers/%s" % (LOOKUPS, e164))
    return str(page.get("country_code") or "").upper()


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--to", action="append", default=[],
                    help="destination in E.164, repeatable")
    ap.add_argument("--media", action="store_true",
                    help="the traffic carries MediaUrl, so a sender must do MMS")
    ap.add_argument("--service", action="append", default=[],
                    help="limit to these Messaging Service SIDs")
    args = ap.parse_args()

    if not args.to:
        log.error("give at least one destination with --to +15551234567")
        return 2

    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)

    destinations = []
    for e164 in args.to:
        country = resolve(session, e164)
        destinations.append({"phone_number": e164, "country_code": country,
                             "needs_mms": args.media})
        log.info("destination %s resolves to %s", e164, country or "?")

    services = list_services(session, 200)
    if args.service:
        services = [s for s in services if s.get("sid") in set(args.service)]

    bad = 0
    for svc in services:
        pool = read_pool(session, svc.get("sid"))
        for dest in destinations:
            state, detail = coverage(pool, dest)
            line = "%-16s %s -> %s  %s" % (state, svc.get("sid"),
                                           dest["phone_number"], detail)
            if state == "covered":
                log.info(line)
                continue
            bad += 1
            log.warning(line)
            log.warning("  repair: POST %s/Services/%s/PhoneNumbers "
                        "PhoneNumberSid=PN... for a %s number%s",
                        MESSAGING, svc.get("sid"), dest["country_code"] or "?",
                        " that is MMS capable" if dest["needs_mms"] else "")

    log.info("%d service(s) x %d destination(s), %d uncovered",
             len(services), len(destinations), bad)
    return 1 if bad else 0


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

/**
 * Report Messaging Services whose sender pool cannot reach a destination.
 *
 * 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 MESSAGING = 'https://messaging.twilio.com/v1';
const LOOKUPS = 'https://lookups.twilio.com/v2';

// Alphanumeric sender IDs are not candidates for these destinations, so a pool
// holding nothing else is uncovered however full it looks.
const ALPHA_EXCLUDED = ['US', 'CA'];

/**
 * Case-insensitive capability test that survives both spellings: the pool
 * returns a list like ['SMS', 'MMS'], the account API returns an object with
 * lowercase keys for the same facts.
 */
export function hasCapability(entry, name) {
  const caps = entry.capabilities ?? [];
  if (!Array.isArray(caps)) return Boolean(caps[name.toLowerCase()]);
  return caps.map((c) => String(c).toLowerCase()).includes(name.toLowerCase());
}

/**
 * Decide whether one sender pool can reach one destination. Pure, so the
 * matching rule can be tested without a network. Returns [state, detail].
 */
export function coverage(pool, destination) {
  const country = String(destination.country_code ?? '').toUpperCase();
  const needsMms = Boolean(destination.needs_mms);
  const numbers = pool.phone_numbers ?? [];
  const codes = pool.short_codes ?? [];
  const alphas = pool.alpha_senders ?? [];

  if (!numbers.length && !codes.length && !alphas.length) {
    return ['no-senders',
      'the pool holds no senders at all, which is 21704 on every send rather ' +
      'than 21703 on this destination.'];
  }
  if (!country) {
    return ['unresolved',
      'the destination country was not resolved, so coverage cannot be ' +
      'decided. Read country_code from Lookup v2 first.'];
  }

  const local = numbers.filter(
    (n) => String(n.country_code ?? '').toUpperCase() === country);
  const localCodes = codes.filter(
    (c) => String(c.country_code ?? '').toUpperCase() === country);

  if (!local.length && !localCodes.length) {
    if (ALPHA_EXCLUDED.includes(country)) {
      return ['unreachable',
        `no ${country} number or short code in the pool. The ${alphas.length} ` +
        `alphanumeric sender(s) do not count: they cannot deliver to ${country}.`];
    }
    if (alphas.length) {
      return ['alpha-only',
        `no ${country} number in the pool, only ${alphas.length} alphanumeric ` +
        'sender(s). They are one way and are not accepted everywhere, so this ' +
        'is deliverable in some countries and 21703 in others.'];
    }
    return ['no-local-sender',
      `no ${country} sender in the pool. Selection may still pick a foreign ` +
      'long code, and this is the shape that returns 21703 when it does not.'];
  }

  if (needsMms && !local.some((n) => hasCapability(n, 'MMS'))) {
    return ['no-mms',
      `${local.length} ${country} sender(s) in the pool and not one of them ` +
      'lists MMS, so any message carrying MediaUrl is 21703 while the text ' +
      'only version of it sends.'];
  }

  const kinds = [];
  if (local.length) kinds.push(`${local.length} number(s)`);
  if (localCodes.length) kinds.push(`${localCodes.length} short code(s)`);
  return ['covered',
    `${kinds.join(', ')} in ${country}${needsMms ? ', MMS capable' : ''}`];
}

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 listServices(auth, limit = 200) {
  let url = `${MESSAGING}/Services`;
  let params = { PageSize: 100 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.services ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

async function readPool(auth, serviceSid) {
  const pool = {};
  for (const [path, key] of [['PhoneNumbers', 'phone_numbers'],
                             ['ShortCodes', 'short_codes'],
                             ['AlphaSenders', 'alpha_senders']]) {
    const page = await get(auth, `${MESSAGING}/Services/${serviceSid}/${path}`,
                           { PageSize: 100 });
    pool[key] = page[key] ?? [];
  }
  return pool;
}

async function resolve(auth, e164) {
  const page = await get(auth, `${LOOKUPS}/PhoneNumbers/${e164}`);
  return String(page.country_code ?? '').toUpperCase();
}

function repeatedFlag(name) {
  const out = [];
  process.argv.forEach((a, i) => { if (a === name) out.push(process.argv[i + 1]); });
  return out.filter(Boolean);
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;
  const tos = repeatedFlag('--to');
  if (!tos.length) {
    console.error('give at least one destination with --to +15551234567');
    process.exitCode = 2;
    return;
  }
  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 needsMms = process.argv.includes('--media');
  const only = new Set(repeatedFlag('--service'));

  const destinations = [];
  for (const e164 of tos) {
    const country = await resolve(auth, e164);
    destinations.push({ phone_number: e164, country_code: country, needs_mms: needsMms });
    console.log(`destination ${e164} resolves to ${country || '?'}`);
  }

  let services = await listServices(auth);
  if (only.size) services = services.filter((s) => only.has(s.sid));

  let bad = 0;
  for (const svc of services) {
    const pool = await readPool(auth, svc.sid);
    for (const dest of destinations) {
      const [state, detail] = coverage(pool, dest);
      const line = `${state.padEnd(16)} ${svc.sid} -> ${dest.phone_number}  ${detail}`;
      if (state === 'covered') { console.log(line); continue; }
      bad += 1;
      console.warn(line);
      console.warn(`  repair: POST ${MESSAGING}/Services/${svc.sid}/PhoneNumbers ` +
                   `PhoneNumberSid=PN... for a ${dest.country_code || '?'} number` +
                   `${dest.needs_mms ? ' that is MMS capable' : ''}`);
    }
  }

  console.log(`${services.length} service(s) x ${destinations.length} ` +
              `destination(s), ${bad} uncovered`);
  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_sender_coverage_audit import coverage, has_capability

US_SMS = {"phone_number": "+12025550100", "country_code": "US", "capabilities": ["SMS"]}
US_MMS = {"phone_number": "+12025550101", "country_code": "US", "capabilities": ["SMS", "MMS"]}
GB_SMS = {"phone_number": "+447700900100", "country_code": "GB", "capabilities": ["SMS"]}
ALPHA = {"sid": "AS1", "alpha_sender": "ACME"}

US = {"country_code": "US", "needs_mms": False}
US_MEDIA = {"country_code": "US", "needs_mms": True}
GB = {"country_code": "GB", "needs_mms": False}


def test_a_us_number_covers_a_us_destination():
    state, detail = coverage({"phone_numbers": [US_SMS]}, US)
    assert state == "covered"
    assert "US" in detail


def test_alpha_senders_do_not_cover_the_us():
    # The whole note: the pool is populated and the destination is unreachable.
    state, detail = coverage({"alpha_senders": [ALPHA, ALPHA, ALPHA]}, US)
    assert state == "unreachable"
    assert "cannot deliver to US" in detail


def test_a_uk_only_pool_cannot_reach_the_us():
    assert coverage({"phone_numbers": [GB_SMS]}, US)[0] == "unreachable"


def test_media_needs_an_mms_capable_sender_in_that_country():
    state, detail = coverage({"phone_numbers": [US_SMS]}, US_MEDIA)
    assert state == "no-mms"
    assert "MediaUrl" in detail
    assert coverage({"phone_numbers": [US_SMS, US_MMS]}, US_MEDIA)[0] == "covered"


def test_an_empty_pool_is_21704_and_says_so():
    state, detail = coverage({}, US)
    assert state == "no-senders"
    assert "21704" in detail


def test_a_short_code_in_the_destination_country_counts():
    pool = {"short_codes": [{"short_code": "12345", "country_code": "US"}]}
    assert coverage(pool, US)[0] == "covered"


def test_a_non_north_american_gap_is_not_reported_as_unreachable():
    # Selection may still pick a foreign long code, so this is a softer state.
    assert coverage({"phone_numbers": [US_SMS]}, GB)[0] == "no-local-sender"
    assert coverage({"phone_numbers": [US_SMS], "alpha_senders": [ALPHA]}, GB)[0] == "alpha-only"


def test_an_unresolved_country_is_never_guessed_at():
    assert coverage({"phone_numbers": [US_SMS]}, {"country_code": ""})[0] == "unresolved"


def test_capabilities_match_across_both_spellings():
    assert has_capability({"capabilities": ["SMS", "MMS"]}, "mms")
    assert has_capability({"capabilities": {"sms": True, "mms": True}}, "MMS")
    assert not has_capability({"capabilities": ["SMS"]}, "MMS")

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { coverage, hasCapability } from './twilio-sender-coverage-audit.mjs';

const US_SMS = { phone_number: '+12025550100', country_code: 'US', capabilities: ['SMS'] };
const US_MMS = { phone_number: '+12025550101', country_code: 'US', capabilities: ['SMS', 'MMS'] };
const GB_SMS = { phone_number: '+447700900100', country_code: 'GB', capabilities: ['SMS'] };
const ALPHA = { sid: 'AS1', alpha_sender: 'ACME' };

const US = { country_code: 'US', needs_mms: false };
const US_MEDIA = { country_code: 'US', needs_mms: true };
const GB = { country_code: 'GB', needs_mms: false };

test('a us number covers a us destination', () => {
  const [state, detail] = coverage({ phone_numbers: [US_SMS] }, US);
  assert.equal(state, 'covered');
  assert.match(detail, /US/);
});

test('alpha senders do not cover the us', () => {
  const [state, detail] = coverage({ alpha_senders: [ALPHA, ALPHA, ALPHA] }, US);
  assert.equal(state, 'unreachable');
  assert.match(detail, /cannot deliver to US/);
});

test('a uk only pool cannot reach the us', () => {
  assert.equal(coverage({ phone_numbers: [GB_SMS] }, US)[0], 'unreachable');
});

test('media needs an mms capable sender in that country', () => {
  const [state, detail] = coverage({ phone_numbers: [US_SMS] }, US_MEDIA);
  assert.equal(state, 'no-mms');
  assert.match(detail, /MediaUrl/);
  assert.equal(coverage({ phone_numbers: [US_SMS, US_MMS] }, US_MEDIA)[0], 'covered');
});

test('an empty pool is 21704 and says so', () => {
  const [state, detail] = coverage({}, US);
  assert.equal(state, 'no-senders');
  assert.match(detail, /21704/);
});

test('a short code in the destination country counts', () => {
  const pool = { short_codes: [{ short_code: '12345', country_code: 'US' }] };
  assert.equal(coverage(pool, US)[0], 'covered');
});

test('a non north american gap is not reported as unreachable', () => {
  assert.equal(coverage({ phone_numbers: [US_SMS] }, GB)[0], 'no-local-sender');
  assert.equal(
    coverage({ phone_numbers: [US_SMS], alpha_senders: [ALPHA] }, GB)[0], 'alpha-only');
});

test('an unresolved country is never guessed at', () => {
  assert.equal(coverage({ phone_numbers: [US_SMS] }, { country_code: '' })[0], 'unresolved');
});

test('capabilities match across both spellings', () => {
  assert.ok(hasCapability({ capabilities: ['SMS', 'MMS'] }, 'mms'));
  assert.ok(hasCapability({ capabilities: { sms: true, mms: true } }, 'MMS'));
  assert.ok(!hasCapability({ capabilities: ['SMS'] }, 'MMS'));
});