21215: dialing permissions block a country you sell into

"""Report Twilio voice dialing permissions that are blocking real traffic.

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 place calls and
spend money.
"""
import argparse
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_dialing_permissions_audit")

HOST = "https://api.twilio.com"
BASE = HOST + "/2010-04-01"
MONITOR = "https://monitor.twilio.com/v1"
VOICE = "https://voice.twilio.com/v1"

# The REST rejection and the TwiML Dial rejection. Same permission, two callers.
BLOCKED_CODES = ("21215", "13227")


def prefix_index(countries):
    """Map every dialling prefix in the countries listing to its ISO codes.

    Pure. The value is a list rather than a single code because prefixes are
    shared: every North American Numbering Plan country answers to 1.
    """
    index = {}
    for c in countries or []:
        iso = str(c.get("iso_code") or "").strip().upper()
        for code in c.get("country_codes") or []:
            digits = str(code or "").strip().lstrip("+")
            if iso and digits.isdigit():
                index.setdefault(digits, set()).add(iso)
    return {k: sorted(v) for k, v in index.items()}


def countries_for(to, index):
    """The ISO codes a destination could belong to, longest prefix first.

    Returns a list, and the list is often longer than one. Picking its first
    member would let this check blame Canada for traffic to the United States,
    so the caller is made to see the ambiguity rather than being handed a guess.
    """
    digits = str(to or "").strip().lstrip("+")
    if not digits.isdigit():
        return []
    for length in range(min(4, len(digits)), 0, -1):
        hit = index.get(digits[:length])
        if hit:
            return list(hit)
    return []


def verdict(country, attempts=0, blocked=0):
    """Decide what one country's permissions are doing to you. Pure.

    attempts is calls seen to that country in the window; blocked is the count
    of 21215/13227 alerts resolved to it. Returns (state, detail).
    """
    iso = str(country.get("iso_code") or "??").strip().upper()
    if country.get("low_risk_numbers_enabled"):
        return ("open",
                "%s is enabled for low risk numbers, so ordinary calls are "
                "permitted. The two high risk switches are separate and are the "
                "subject of the companion check." % iso)

    if blocked:
        return ("blocking-live-traffic",
                "%s has low_risk_numbers_enabled false and %d call(s) were "
                "refused with 21215 or 13227 in this window. This is an outage "
                "in a country you are selling into." % (iso, blocked))

    if attempts:
        return ("blocking-attempted",
                "%s has low_risk_numbers_enabled false and %d call(s) were "
                "placed toward it. No refusal alert landed in this window, so "
                "check the window before concluding they got through."
                % (iso, attempts))

    return ("closed-unused",
            "%s is disabled and nothing was dialled toward it. Almost every "
            "account looks like this for almost every country; it is context, "
            "not a finding." % iso)


def settings_verdict(settings, subaccounts=0):
    """Decide whether subaccounts get the parent's permissions at all. Pure.

    Returns (state, detail). This is the check that explains a regression with
    no deploy behind it: the same code on a subaccount, refused.
    """
    if settings.get("dialing_permissions_inheritance"):
        return ("inherited",
                "dialing_permissions_inheritance is true, so subaccounts use "
                "the parent's country permissions.")
    if subaccounts:
        return ("not-inherited",
                "dialing_permissions_inheritance is false and this account has "
                "%d subaccount(s). Each one carries its own home-country-only "
                "default, so enabling a country here does nothing for them."
                % subaccounts)
    return ("not-inherited-no-subaccounts",
            "dialing_permissions_inheritance is false, which changes nothing "
            "today because there are no subaccounts. It will change everything "
            "on the day somebody creates one.")


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 page_meta(session, url, key, **params):
    """Page an API that paginates with an absolute meta.next_page_url."""
    params.setdefault("PageSize", 1000)
    out = []
    while url:
        page = get(session, url, **params)
        out.extend(page.get(key, []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out


def page_2010(session, url, key, **params):
    """Page a 2010-04-01 listing. next_page_uri here is a path, not a URL."""
    params.setdefault("PageSize", 1000)
    out = []
    while url:
        body = get(session, url, **params)
        out.extend(body.get(key, []))
        nxt = body.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out


def sweep_alerts(session, since, limit):
    """Alerts at both log levels, merged on sid.

    Several voice failures are logged at warning rather than error, so an
    error-only sweep reports a clean account while the calls keep failing.
    """
    seen = {}
    for level in ("error", "warning"):
        url = MONITOR + "/Alerts"
        params = {"LogLevel": level, "StartDate": since, "PageSize": 1000}
        count = 0
        while url and count < limit:
            page = get(session, url, **params)
            for a in page.get("alerts", []):
                seen.setdefault(a.get("sid"), a)
                count += 1
            url = (page.get("meta") or {}).get("next_page_url")
            params = {}
    return list(seen.values())


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=7,
                    help="window to sweep (alerts are retained 30 days)")
    ap.add_argument("--max-calls", type=int, default=20000,
                    help="stop after this many calls when counting attempts")
    ap.add_argument("--no-calls", action="store_true",
                    help="skip the Calls listing and rely on alerts alone")
    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)

    since = (dt.date.today() - dt.timedelta(days=min(args.days, 30))).isoformat()
    countries = page_meta(session, VOICE + "/DialingPermissions/Countries",
                          "content")
    if not countries:
        log.info("no dialing permission countries returned")
        return 0
    index = prefix_index(countries)

    attempts = {}
    if not args.no_calls:
        for c in page_2010(session, "%s/Accounts/%s/Calls.json" % (BASE, account),
                           "calls", **{"StartTime>=": since}):
            for iso in countries_for(c.get("to"), index):
                attempts[iso] = attempts.get(iso, 0) + 1

    blocked = {}
    calls = {}
    for a in sweep_alerts(session, since, 10000):
        if str(a.get("error_code") or "").strip() not in BLOCKED_CODES:
            continue
        sid = str(a.get("resource_sid") or "")
        if not sid.startswith("CA"):
            continue
        if sid not in calls:
            calls[sid] = get(session, "%s/Accounts/%s/Calls/%s.json"
                             % (BASE, account, sid))
        for iso in countries_for(calls[sid].get("to"), index):
            blocked[iso] = blocked.get(iso, 0) + 1

    findings = 0
    for c in sorted(countries, key=lambda x: str(x.get("iso_code") or "")):
        iso = str(c.get("iso_code") or "").strip().upper()
        state, detail = verdict(c, attempts.get(iso, 0), blocked.get(iso, 0))
        if state in ("open", "closed-unused"):
            continue
        findings += 1
        log.warning("%-22s %s", state, detail)

    subaccounts = len(page_2010(session, BASE + "/Accounts.json", "accounts")) - 1
    state, detail = settings_verdict(get(session, VOICE + "/Settings"),
                                     max(subaccounts, 0))
    (log.info if state == "inherited" else log.warning)("%-22s %s", state, detail)

    log.info("%d blocked destination(s) with traffic across %d country entries",
             findings, len(countries))
    if findings or state == "not-inherited":
        log.warning("  repair: POST %s/DialingPermissions/BulkCountryUpdates "
                    "with an UpdateRequest array of "
                    "{\"iso_code\":\"XX\",\"low_risk_numbers_enabled\":true}", VOICE)
        log.warning("  repair: POST %s/Settings with "
                    "DialingPermissionsInheritance=true to stop every new "
                    "subaccount starting from the home-country default", VOICE)
        return 1
    return 0


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

/**
 * Report Twilio voice dialing permissions that are blocking real traffic.
 *
 * 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 MONITOR = 'https://monitor.twilio.com/v1';
const VOICE = 'https://voice.twilio.com/v1';

// The REST rejection and the TwiML Dial rejection. Same permission, two callers.
const BLOCKED_CODES = ['21215', '13227'];

/**
 * Map every dialling prefix in the countries listing to its ISO codes. Pure.
 * The value is a list because prefixes are shared: every NANP country answers
 * to 1.
 */
export function prefixIndex(countries) {
  const index = new Map();
  for (const c of countries ?? []) {
    const iso = String(c.iso_code ?? '').trim().toUpperCase();
    for (const code of c.country_codes ?? []) {
      const digits = String(code ?? '').trim().replace(/^\+/, '');
      if (!iso || !/^[0-9]+$/.test(digits)) continue;
      if (!index.has(digits)) index.set(digits, new Set());
      index.get(digits).add(iso);
    }
  }
  return Object.fromEntries([...index].map(([k, v]) => [k, [...v].sort()]));
}

/**
 * The ISO codes a destination could belong to, longest prefix first. Returns a
 * list, and the list is often longer than one: picking its first member would
 * let this check blame Canada for traffic to the United States.
 */
export function countriesFor(to, index) {
  const digits = String(to ?? '').trim().replace(/^\+/, '');
  if (!/^[0-9]+$/.test(digits)) return [];
  for (let length = Math.min(4, digits.length); length > 0; length -= 1) {
    const hit = index[digits.slice(0, length)];
    if (hit) return [...hit];
  }
  return [];
}

/**
 * Decide what one country's permissions are doing to you. Pure. Returns
 * [state, detail].
 */
export function verdict(country, attempts = 0, blocked = 0) {
  const iso = String(country.iso_code ?? '??').trim().toUpperCase();
  if (country.low_risk_numbers_enabled) {
    return ['open',
      `${iso} is enabled for low risk numbers, so ordinary calls are permitted. ` +
      'The two high risk switches are separate and are the subject of the ' +
      'companion check.'];
  }

  if (blocked) {
    return ['blocking-live-traffic',
      `${iso} has low_risk_numbers_enabled false and ${blocked} call(s) were ` +
      'refused with 21215 or 13227 in this window. This is an outage in a ' +
      'country you are selling into.'];
  }

  if (attempts) {
    return ['blocking-attempted',
      `${iso} has low_risk_numbers_enabled false and ${attempts} call(s) were ` +
      'placed toward it. No refusal alert landed in this window, so check the ' +
      'window before concluding they got through.'];
  }

  return ['closed-unused',
    `${iso} is disabled and nothing was dialled toward it. Almost every account ` +
    'looks like this for almost every country; it is context, not a finding.'];
}

/**
 * Decide whether subaccounts get the parent's permissions at all. Pure. This is
 * the check that explains a regression with no deploy behind it.
 */
export function settingsVerdict(settings, subaccounts = 0) {
  if (settings.dialing_permissions_inheritance) {
    return ['inherited',
      'dialing_permissions_inheritance is true, so subaccounts use the ' +
      "parent's country permissions."];
  }
  if (subaccounts) {
    return ['not-inherited',
      `dialing_permissions_inheritance is false and this account has ` +
      `${subaccounts} subaccount(s). Each one carries its own ` +
      'home-country-only default, so enabling a country here does nothing for them.'];
  }
  return ['not-inherited-no-subaccounts',
    'dialing_permissions_inheritance is false, which changes nothing today ' +
    'because there are no subaccounts. It will change everything on the day ' +
    'somebody creates one.'];
}

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

async function pageMeta(auth, url, key, params = {}) {
  const out = [];
  let next = url;
  let p = { PageSize: 1000, ...params };
  while (next) {
    const page = await get(auth, next, p);
    out.push(...(page[key] ?? []));
    next = page.meta?.next_page_url ?? null;
    p = {};
  }
  return out;
}

async function page2010(auth, url, key, params = {}) {
  const out = [];
  let next = url;
  let p = { PageSize: 1000, ...params };
  while (next) {
    const body = await get(auth, next, p);
    out.push(...(body[key] ?? []));
    next = body.next_page_uri ? HOST + body.next_page_uri : null;
    p = {};
  }
  return out;
}

async function sweepAlerts(auth, since, limit = 10000) {
  const seen = new Map();
  for (const level of ['error', 'warning']) {
    let url = `${MONITOR}/Alerts`;
    let p = { LogLevel: level, StartDate: since, PageSize: 1000 };
    let count = 0;
    while (url && count < limit) {
      const page = await get(auth, url, p);
      for (const a of page.alerts ?? []) {
        if (!seen.has(a.sid)) seen.set(a.sid, a);
        count += 1;
      }
      url = page.meta?.next_page_url ?? null;
      p = {};
    }
  }
  return [...seen.values()];
}

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 i = process.argv.indexOf('--days');
  const days = Math.min(i === -1 ? 7 : Number(process.argv[i + 1]), 30);
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);

  const countries = await pageMeta(auth, `${VOICE}/DialingPermissions/Countries`, 'content');
  if (countries.length === 0) {
    console.log('no dialing permission countries returned');
    return;
  }
  const index = prefixIndex(countries);

  const attempts = new Map();
  if (!process.argv.includes('--no-calls')) {
    const calls = await page2010(auth, `${BASE}/Accounts/${account}/Calls.json`,
                                 'calls', { 'StartTime>=': since });
    for (const c of calls) {
      for (const iso of countriesFor(c.to, index)) {
        attempts.set(iso, (attempts.get(iso) ?? 0) + 1);
      }
    }
  }

  const blocked = new Map();
  const seen = new Map();
  for (const a of await sweepAlerts(auth, since)) {
    if (!BLOCKED_CODES.includes(String(a.error_code ?? '').trim())) continue;
    const sid = String(a.resource_sid ?? '');
    if (!sid.startsWith('CA')) continue;
    if (!seen.has(sid)) {
      seen.set(sid, await get(auth, `${BASE}/Accounts/${account}/Calls/${sid}.json`));
    }
    for (const iso of countriesFor(seen.get(sid).to, index)) {
      blocked.set(iso, (blocked.get(iso) ?? 0) + 1);
    }
  }

  let findings = 0;
  for (const c of [...countries].sort((a, b) =>
    String(a.iso_code ?? '').localeCompare(String(b.iso_code ?? '')))) {
    const iso = String(c.iso_code ?? '').trim().toUpperCase();
    const [state, detail] = verdict(c, attempts.get(iso) ?? 0, blocked.get(iso) ?? 0);
    if (state === 'open' || state === 'closed-unused') continue;
    findings += 1;
    console.warn(`${state.padEnd(22)} ${detail}`);
  }

  const accounts = await page2010(auth, `${BASE}/Accounts.json`, 'accounts');
  const [state, detail] = settingsVerdict(await get(auth, `${VOICE}/Settings`),
                                          Math.max(accounts.length - 1, 0));
  (state === 'inherited' ? console.log : console.warn)(`${state.padEnd(22)} ${detail}`);

  console.log(`${findings} blocked destination(s) with traffic across ` +
              `${countries.length} country entries`);
  if (findings || state === 'not-inherited') {
    console.warn(`  repair: POST ${VOICE}/DialingPermissions/BulkCountryUpdates ` +
                 'with an UpdateRequest array of ' +
                 '{"iso_code":"XX","low_risk_numbers_enabled":true}');
    console.warn(`  repair: POST ${VOICE}/Settings with ` +
                 'DialingPermissionsInheritance=true to stop every new subaccount ' +
                 'starting from the home-country default');
    process.exitCode = 1;
  }
}

// Only run when invoked directly, so importing this module from the test file
// does not fire main(), fail on the missing credentials and set an exit code
// that fails the suite 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_dialing_permissions_audit import (countries_for, prefix_index,
                                                 settings_verdict, verdict)

LISTING = [
    {"iso_code": "US", "country_codes": ["1"], "low_risk_numbers_enabled": True},
    {"iso_code": "CA", "country_codes": ["1"], "low_risk_numbers_enabled": True},
    {"iso_code": "GB", "country_codes": ["44"], "low_risk_numbers_enabled": False},
    {"iso_code": "AU", "country_codes": ["61"], "low_risk_numbers_enabled": False},
]


def test_shared_dialling_code_resolves_to_the_whole_group():
    # Picking one member would blame Canada for traffic to the United States.
    assert countries_for("+14155550100", prefix_index(LISTING)) == ["CA", "US"]


def test_longest_prefix_wins():
    index = prefix_index([{"iso_code": "GB", "country_codes": ["44"]},
                          {"iso_code": "XX", "country_codes": ["4470"]}])
    assert countries_for("+447012345678", index) == ["XX"]


def test_destination_outside_every_prefix_resolves_to_nothing():
    assert countries_for("not-a-number", prefix_index(LISTING)) == []


def test_disabled_country_with_refusals_is_an_outage():
    state, detail = verdict(LISTING[2], attempts=40, blocked=12)
    assert state == "blocking-live-traffic"
    assert "21215" in detail


def test_disabled_country_with_traffic_but_no_alerts_is_softer():
    assert verdict(LISTING[2], attempts=40)[0] == "blocking-attempted"


def test_disabled_country_nobody_calls_is_context_not_a_finding():
    state, detail = verdict(LISTING[3])
    assert state == "closed-unused"
    assert "not a finding" in detail


def test_inheritance_off_with_subaccounts_explains_the_regression():
    state, detail = settings_verdict({"dialing_permissions_inheritance": False}, 6)
    assert state == "not-inherited"
    assert "6 subaccount(s)" in detail


def test_inheritance_off_without_subaccounts_is_a_future_problem():
    assert settings_verdict({"dialing_permissions_inheritance": False})[0] == \
        "not-inherited-no-subaccounts"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { countriesFor, prefixIndex, settingsVerdict, verdict }
  from './twilio-dialing-permissions-audit.mjs';

const LISTING = [
  { iso_code: 'US', country_codes: ['1'], low_risk_numbers_enabled: true },
  { iso_code: 'CA', country_codes: ['1'], low_risk_numbers_enabled: true },
  { iso_code: 'GB', country_codes: ['44'], low_risk_numbers_enabled: false },
  { iso_code: 'AU', country_codes: ['61'], low_risk_numbers_enabled: false },
];

test('shared dialling code resolves to the whole group', () => {
  assert.deepEqual(countriesFor('+14155550100', prefixIndex(LISTING)), ['CA', 'US']);
});

test('longest prefix wins', () => {
  const index = prefixIndex([{ iso_code: 'GB', country_codes: ['44'] },
                             { iso_code: 'XX', country_codes: ['4470'] }]);
  assert.deepEqual(countriesFor('+447012345678', index), ['XX']);
});

test('destination outside every prefix resolves to nothing', () => {
  assert.deepEqual(countriesFor('not-a-number', prefixIndex(LISTING)), []);
});

test('disabled country with refusals is an outage', () => {
  const [state, detail] = verdict(LISTING[2], 40, 12);
  assert.equal(state, 'blocking-live-traffic');
  assert.match(detail, /21215/);
});

test('disabled country with traffic but no alerts is softer', () => {
  assert.equal(verdict(LISTING[2], 40)[0], 'blocking-attempted');
});

test('disabled country nobody calls is context not a finding', () => {
  const [state, detail] = verdict(LISTING[3]);
  assert.equal(state, 'closed-unused');
  assert.match(detail, /not a finding/);
});

test('inheritance off with subaccounts explains the regression', () => {
  const [state, detail] = settingsVerdict({ dialing_permissions_inheritance: false }, 6);
  assert.equal(state, 'not-inherited');
  assert.match(detail, /6 subaccount\(s\)/);
});

test('inheritance off without subaccounts is a future problem', () => {
  assert.equal(settingsVerdict({ dialing_permissions_inheritance: false })[0],
               'not-inherited-no-subaccounts');
});