two toll-free numbers in one sender pool get both blocked

"""Report Messaging Services holding more than one toll-free sender.

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 collections
import datetime as dt
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_pool_audit")

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

# North American toll-free area codes. Matching the numbering plan rather than a
# prefix string keeps a UK freephone number and a subscriber number containing
# 800 out of the report.
TOLL_FREE_AREA_CODES = frozenset({"800", "833", "844", "855", "866", "877", "888"})

TOLL_FREE_ERROR = "30032"


def is_toll_free(phone_number):
    """True for a North American toll-free number in any formatting.

    Pure. Eleven digits beginning with the country code 1, and an area code from
    the toll-free set: that is the rule, and it is deliberately not a substring
    test on the E.164 string.
    """
    digits = "".join(c for c in str(phone_number or "") if c.isdigit())
    if len(digits) != 11 or not digits.startswith("1"):
        return False
    return digits[1:4] in TOLL_FREE_AREA_CODES


def verdict(entries):
    """Classify one sender pool by how many toll-free numbers share it.

    `entries` is the pool's phone_numbers list. Pure, so the rule is testable
    without a network. Returns (state, detail).
    """
    entries = entries or []
    if not entries:
        return ("empty", "no phone numbers in this pool at all, which is 21704 "
                         "on every send and a different note.")

    toll_free = [str(e.get("phone_number") or "")
                 for e in entries if is_toll_free(e.get("phone_number"))]
    others = len(entries) - len(toll_free)

    if not toll_free:
        return ("no-toll-free",
                "%d sender(s), none of them toll-free." % len(entries))
    if len(toll_free) == 1:
        return ("single-toll-free",
                "one toll-free sender (%s) alongside %d other sender(s), which "
                "is the shape Twilio's guidance asks for."
                % (toll_free[0], others))
    return ("multiple-toll-free",
            "%d toll-free senders share this pool: %s. Carriers read that as "
            "snowshoeing and block the numbers, including ones verified long "
            "before the extras were added."
            % (len(toll_free), ", ".join(toll_free)))


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


def count_blocks(session, account, days, max_messages):
    """Count 30032 per sender. The Messages list has no error code filter and no
    status filter, so the window is paged and filtered here."""
    since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    url = "%s/Accounts/%s/Messages.json" % (BASE, account)
    params = {"DateSent>": since, "PageSize": 1000}
    seen = 0
    tally = collections.Counter()
    while url and seen < max_messages:
        page = get(session, url, **params)
        rows = page.get("messages", [])
        seen += len(rows)
        for m in rows:
            if str(m.get("error_code") or "") == TOLL_FREE_ERROR:
                tally[m.get("from")] += 1
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return tally


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-services", type=int, default=200,
                    help="stop paging after this many Messaging Services")
    ap.add_argument("--check-errors", action="store_true",
                    help="page the Messages list and count 30032 per sender")
    ap.add_argument("--days", type=int, default=7,
                    help="window in days for --check-errors")
    ap.add_argument("--max-messages", type=int, default=20000,
                    help="stop paging the Messages list after this many rows")
    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)

    blocks = count_blocks(session, account, args.days, args.max_messages) if args.check_errors else {}

    services = list_services(session, args.max_services)
    bad = 0
    for svc in services:
        entries = read_pool(session, svc.get("sid"))
        state, detail = verdict(entries)
        label = svc.get("friendly_name") or svc.get("sid")
        line = "%-19s %s  %s" % (state, label, detail)
        if state != "multiple-toll-free":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        for e in entries:
            number = e.get("phone_number")
            if is_toll_free(number) and blocks.get(number):
                log.warning("  %s has %d message(s) failing %s in the last %d day(s)",
                            number, blocks[number], TOLL_FREE_ERROR, args.days)
        log.warning("  repair: give each toll-free number its own Messaging "
                    "Service, then DELETE %s/Services/%s/PhoneNumbers/{PNSid} "
                    "for the extras and point each traffic stream at the right "
                    "MessagingServiceSid.", MESSAGING, svc.get("sid"))

    log.info("%d service(s), %d holding more than one toll-free sender",
             len(services), bad)
    return 1 if bad else 0


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

/**
 * Report Messaging Services holding more than one toll-free sender.
 *
 * 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 MESSAGING = 'https://messaging.twilio.com/v1';

// North American toll-free area codes. Matching the numbering plan rather than a
// prefix string keeps a UK freephone number and a subscriber number containing
// 800 out of the report.
const TOLL_FREE_AREA_CODES = new Set(['800', '833', '844', '855', '866', '877', '888']);

const TOLL_FREE_ERROR = '30032';

/**
 * True for a North American toll-free number in any formatting. Pure: eleven
 * digits beginning with country code 1, and an area code from the toll-free set.
 */
export function isTollFree(phoneNumber) {
  const digits = String(phoneNumber ?? '').replace(/\D/g, '');
  if (digits.length !== 11 || !digits.startsWith('1')) return false;
  return TOLL_FREE_AREA_CODES.has(digits.slice(1, 4));
}

/**
 * Classify one sender pool by how many toll-free numbers share it. Pure, so the
 * rule is testable without a network. Returns [state, detail].
 */
export function verdict(entries) {
  const pool = entries ?? [];
  if (!pool.length) {
    return ['empty',
      'no phone numbers in this pool at all, which is 21704 on every send and ' +
      'a different note.'];
  }

  const tollFree = pool.filter((e) => isTollFree(e.phone_number))
                       .map((e) => String(e.phone_number ?? ''));
  const others = pool.length - tollFree.length;

  if (!tollFree.length) {
    return ['no-toll-free', `${pool.length} sender(s), none of them toll-free.`];
  }
  if (tollFree.length === 1) {
    return ['single-toll-free',
      `one toll-free sender (${tollFree[0]}) alongside ${others} other ` +
      "sender(s), which is the shape Twilio's guidance asks for."];
  }
  return ['multiple-toll-free',
    `${tollFree.length} toll-free senders share this pool: ${tollFree.join(', ')}. ` +
    'Carriers read that as snowshoeing and block the numbers, including ones ' +
    'verified long before the extras were added.'];
}

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) {
  let url = `${MESSAGING}/Services/${serviceSid}/PhoneNumbers`;
  let params = { PageSize: 100 };
  const out = [];
  while (url) {
    const page = await get(auth, url, params);
    out.push(...(page.phone_numbers ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out;
}

/** Count 30032 per sender. The Messages list has no error code filter. */
async function countBlocks(auth, account, days, maxMessages) {
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);
  let url = `${BASE}/Accounts/${account}/Messages.json`;
  let params = { 'DateSent>': since, PageSize: 1000 };
  const tally = new Map();
  let seen = 0;
  while (url && seen < maxMessages) {
    const page = await get(auth, url, params);
    const rows = page.messages ?? [];
    seen += rows.length;
    for (const m of rows) {
      if (String(m.error_code ?? '') === TOLL_FREE_ERROR) {
        tally.set(m.from, (tally.get(m.from) ?? 0) + 1);
      }
    }
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return tally;
}

function flagValue(name, fallback) {
  const i = process.argv.indexOf(name);
  return i === -1 ? fallback : Number(process.argv[i + 1]);
}

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 days = flagValue('--days', 7);
  const blocks = process.argv.includes('--check-errors')
    ? await countBlocks(auth, account, days, flagValue('--max-messages', 20000))
    : new Map();

  const services = await listServices(auth);
  let bad = 0;
  for (const svc of services) {
    const entries = await readPool(auth, svc.sid);
    const [state, detail] = verdict(entries);
    const label = svc.friendly_name ?? svc.sid;
    const line = `${state.padEnd(19)} ${label}  ${detail}`;
    if (state !== 'multiple-toll-free') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    for (const e of entries) {
      if (isTollFree(e.phone_number) && blocks.get(e.phone_number)) {
        console.warn(`  ${e.phone_number} has ${blocks.get(e.phone_number)} ` +
                     `message(s) failing ${TOLL_FREE_ERROR} in the last ${days} day(s)`);
      }
    }
    console.warn('  repair: give each toll-free number its own Messaging Service, ' +
                 `then DELETE ${MESSAGING}/Services/${svc.sid}/PhoneNumbers/{PNSid} ` +
                 'for the extras and point each traffic stream at the right ' +
                 'MessagingServiceSid.');
  }

  console.log(`${services.length} service(s), ${bad} holding more than one toll-free sender`);
  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_pool_audit import is_toll_free, verdict


def pn(number):
    return {"phone_number": number, "sid": "PN" + number[-4:]}


def test_every_toll_free_area_code_is_recognised():
    for area in ("800", "833", "844", "855", "866", "877", "888"):
        assert is_toll_free("+1" + area + "5550123"), area


def test_a_uk_freephone_number_is_not_north_american_toll_free():
    # +44 800 is freephone in the UK and nothing to do with this rule.
    assert not is_toll_free("+448001234567")


def test_a_subscriber_number_containing_800_is_not_toll_free():
    assert not is_toll_free("+12028005550")
    assert not is_toll_free("+15558675309")


def test_formatting_does_not_change_the_answer():
    assert is_toll_free("+1 (833) 555-0123")
    assert is_toll_free("18335550123")


def test_one_toll_free_number_is_the_recommended_shape():
    state, detail = verdict([pn("+18005550123"), pn("+12025550100")])
    assert state == "single-toll-free"
    assert "+18005550123" in detail


def test_two_toll_free_numbers_in_one_pool_is_the_finding():
    state, detail = verdict([pn("+18005550123"), pn("+18445550199")])
    assert state == "multiple-toll-free"
    assert "snowshoeing" in detail
    assert "+18445550199" in detail


def test_a_pool_of_long_codes_is_not_this_note():
    assert verdict([pn("+12025550100"), pn("+12025550101")])[0] == "no-toll-free"


def test_an_empty_pool_points_at_the_other_note():
    state, detail = verdict([])
    assert state == "empty"
    assert "21704" in detail

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

const pn = (number) => ({ phone_number: number, sid: `PN${number.slice(-4)}` });

test('every toll free area code is recognised', () => {
  for (const area of ['800', '833', '844', '855', '866', '877', '888']) {
    assert.ok(isTollFree(`+1${area}5550123`), area);
  }
});

test('a uk freephone number is not north american toll free', () => {
  assert.ok(!isTollFree('+448001234567'));
});

test('a subscriber number containing 800 is not toll free', () => {
  assert.ok(!isTollFree('+12028005550'));
  assert.ok(!isTollFree('+15558675309'));
});

test('formatting does not change the answer', () => {
  assert.ok(isTollFree('+1 (833) 555-0123'));
  assert.ok(isTollFree('18335550123'));
});

test('one toll free number is the recommended shape', () => {
  const [state, detail] = verdict([pn('+18005550123'), pn('+12025550100')]);
  assert.equal(state, 'single-toll-free');
  assert.match(detail, /\+18005550123/);
});

test('two toll free numbers in one pool is the finding', () => {
  const [state, detail] = verdict([pn('+18005550123'), pn('+18445550199')]);
  assert.equal(state, 'multiple-toll-free');
  assert.match(detail, /snowshoeing/);
  assert.match(detail, /\+18445550199/);
});

test('a pool of long codes is not this note', () => {
  assert.equal(verdict([pn('+12025550100'), pn('+12025550101')])[0], 'no-toll-free');
});

test('an empty pool points at the other note', () => {
  const [state, detail] = verdict([]);
  assert.equal(state, 'empty');
  assert.match(detail, /21704/);
});