high risk dialing prefixes left open to toll fraud

"""Report Twilio countries whose high risk dialing classes are left open.

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

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


def money(price):
    """A Twilio price as a positive amount.

    Prices arrive as strings and outbound ones are negative, because they are
    charges against the account. Absolute value here so the report reads as
    money spent rather than as a balance, and 0.0 on anything unparseable so a
    missing price never takes the run down.
    """
    try:
        return abs(float(str(price or "0").strip() or 0))
    except ValueError:
        return 0.0


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

    Pure. The value is a list 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."""
    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, served=(), attempts=0, spend=0.0):
    """Classify one country's high risk exposure. Pure, so the rules can be
    tested without a network.

    served is the set of ISO codes the business actually calls into; it has to
    be declared because no API can infer it. Returns (state, detail).
    """
    iso = str(country.get("iso_code") or "??").strip().upper()
    serving = {str(s).strip().upper() for s in served}
    special = bool(country.get("high_risk_special_numbers_enabled"))
    fraud = bool(country.get("high_risk_tollfraud_numbers_enabled"))
    low = bool(country.get("low_risk_numbers_enabled"))

    if not (special or fraud):
        return ("closed",
                "%s has both high risk classes disabled, so its premium and "
                "toll fraud ranges are not reachable from this account." % iso)

    classes = ", ".join([n for n, on in
                         (("high_risk_special_numbers_enabled", special),
                          ("high_risk_tollfraud_numbers_enabled", fraud)) if on])

    if attempts:
        return ("open-and-dialled",
                "%s has %s and %d call(s) already went to it in this window, "
                "costing %.2f. This has stopped being a risk assessment: check "
                "what placed them before you close anything."
                % (iso, classes, attempts, spend))

    if not low:
        return ("premium-only",
                "%s has low_risk_numbers_enabled false while %s is true: an "
                "ordinary business call to this country is refused and its most "
                "expensive ranges are not. Nobody configures that deliberately."
                % (iso, classes))

    if iso in serving:
        return ("open-in-market",
                "%s is a country you serve and %s is on. Low risk traffic is "
                "what your customers are; the high risk classes are what "
                "somebody else's revenue share is." % (iso, classes))

    return ("open-unused",
            "%s is outside the served set and %s is on. This is exposure "
            "carried for no return, in exactly the kind of country an IRSF "
            "range sits in." % (iso, classes))


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, limit, **params):
    """Page a 2010-04-01 listing. next_page_uri here is a path, not a URL."""
    params.setdefault("PageSize", 1000)
    out = []
    while url and len(out) < limit:
        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[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--serve", default="",
                    help="comma separated ISO codes your business calls into")
    ap.add_argument("--days", type=int, default=30,
                    help="window over which to count traffic and spend")
    ap.add_argument("--max-calls", type=int, default=20000,
                    help="stop after this many calls")
    ap.add_argument("--no-calls", action="store_true",
                    help="skip the traffic join and report permissions 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)

    served = [s for s in (p.strip().upper() for p in args.serve.split(",")) if s]
    if not served:
        log.warning("no --serve list given: every country with a high risk "
                    "class open will be reported as unused")

    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, spend = {}, {}
    if not args.no_calls:
        since = (dt.date.today() - dt.timedelta(days=args.days)).isoformat()
        calls = page_2010(session, "%s/Accounts/%s/Calls.json" % (BASE, account),
                          "calls", args.max_calls, **{"StartTime>=": since})
        for c in calls:
            for iso in countries_for(c.get("to"), index):
                attempts[iso] = attempts.get(iso, 0) + 1
                spend[iso] = spend.get(iso, 0.0) + money(c.get("price"))

    findings = []
    for c in countries:
        iso = str(c.get("iso_code") or "").strip().upper()
        state, detail = verdict(c, served, attempts.get(iso, 0), spend.get(iso, 0.0))
        if state == "closed":
            continue
        findings.append((state, iso, detail))

    order = {"open-and-dialled": 0, "premium-only": 1, "open-unused": 2,
             "open-in-market": 3}
    for state, iso, detail in sorted(findings, key=lambda f: (order.get(f[0], 9), f[1])):
        log.warning("%-17s %s", state, detail)

    unserved = [f for f in findings if f[0] in ("open-unused", "premium-only",
                                                "open-and-dialled")]
    log.info("%d country entries with a high risk class open outside the served set",
             len(unserved))
    if not findings:
        return 0
    log.warning("  repair: POST %s/DialingPermissions/BulkCountryUpdates with an "
                "UpdateRequest array disabling high_risk_special_numbers_enabled "
                "and high_risk_tollfraud_numbers_enabled for every unused ISO "
                "code", VOICE)
    log.warning("  repair: run this on a schedule. Permissions get widened "
                "during incidents and the widening outlives the incident")
    return 1 if unserved else 0


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

/**
 * Report Twilio countries whose high risk dialing classes are left open.
 *
 * 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 VOICE = 'https://voice.twilio.com/v1';

/**
 * A Twilio price as a positive amount. Prices arrive as strings and outbound
 * ones are negative, because they are charges against the account. Zero on
 * anything unparseable so a missing price never takes the run down.
 */
export function money(price) {
  const n = Number.parseFloat(String(price ?? '0').trim());
  return Number.isFinite(n) ? Math.abs(n) : 0;
}

/** Map every dialling prefix in the countries listing to its ISO codes. Pure. */
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. */
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 [];
}

/**
 * Classify one country's high risk exposure. Pure. `served` is the set of ISO
 * codes the business actually calls into, which has to be declared because no
 * API can infer it. Returns [state, detail].
 */
export function verdict(country, served = [], attempts = 0, spend = 0) {
  const iso = String(country.iso_code ?? '??').trim().toUpperCase();
  const serving = new Set([...served].map((s) => String(s).trim().toUpperCase()));
  const special = Boolean(country.high_risk_special_numbers_enabled);
  const fraud = Boolean(country.high_risk_tollfraud_numbers_enabled);
  const low = Boolean(country.low_risk_numbers_enabled);

  if (!special && !fraud) {
    return ['closed',
      `${iso} has both high risk classes disabled, so its premium and toll ` +
      'fraud ranges are not reachable from this account.'];
  }

  const classes = [['high_risk_special_numbers_enabled', special],
                   ['high_risk_tollfraud_numbers_enabled', fraud]]
    .filter(([, on]) => on).map(([n]) => n).join(', ');

  if (attempts) {
    return ['open-and-dialled',
      `${iso} has ${classes} and ${attempts} call(s) already went to it in this ` +
      `window, costing ${spend.toFixed(2)}. This has stopped being a risk ` +
      'assessment: check what placed them before you close anything.'];
  }

  if (!low) {
    return ['premium-only',
      `${iso} has low_risk_numbers_enabled false while ${classes} is true: an ` +
      'ordinary business call to this country is refused and its most expensive ' +
      'ranges are not. Nobody configures that deliberately.'];
  }

  if (serving.has(iso)) {
    return ['open-in-market',
      `${iso} is a country you serve and ${classes} is on. Low risk traffic is ` +
      "what your customers are; the high risk classes are what somebody else's " +
      'revenue share is.'];
  }

  return ['open-unused',
    `${iso} is outside the served set and ${classes} is on. This is exposure ` +
    'carried for no return, in exactly the kind of country an IRSF range sits in.'];
}

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, limit, params = {}) {
  const out = [];
  let next = url;
  let p = { PageSize: 1000, ...params };
  while (next && out.length < limit) {
    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.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 flag = (name, fallback) => {
    const i = process.argv.indexOf(name);
    return i === -1 ? fallback : process.argv[i + 1];
  };
  const served = String(flag('--serve', '')).split(',')
    .map((s) => s.trim().toUpperCase()).filter(Boolean);
  const days = Number(flag('--days', 30));
  if (served.length === 0) {
    console.warn('no --serve list given: every country with a high risk class ' +
                 'open will be reported as unused');
  }

  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();
  const spend = new Map();
  if (!process.argv.includes('--no-calls')) {
    const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);
    const calls = await page2010(auth, `${BASE}/Accounts/${account}/Calls.json`,
                                 'calls', 20000, { 'StartTime>=': since });
    for (const c of calls) {
      for (const iso of countriesFor(c.to, index)) {
        attempts.set(iso, (attempts.get(iso) ?? 0) + 1);
        spend.set(iso, (spend.get(iso) ?? 0) + money(c.price));
      }
    }
  }

  const findings = [];
  for (const c of countries) {
    const iso = String(c.iso_code ?? '').trim().toUpperCase();
    const [state, detail] = verdict(c, served, attempts.get(iso) ?? 0,
                                    spend.get(iso) ?? 0);
    if (state === 'closed') continue;
    findings.push([state, iso, detail]);
  }

  const order = { 'open-and-dialled': 0, 'premium-only': 1, 'open-unused': 2,
                  'open-in-market': 3 };
  findings.sort((a, b) => (order[a[0]] ?? 9) - (order[b[0]] ?? 9) ||
                          a[1].localeCompare(b[1]));
  for (const [state, , detail] of findings) {
    console.warn(`${state.padEnd(17)} ${detail}`);
  }

  const unserved = findings.filter(([s]) =>
    ['open-unused', 'premium-only', 'open-and-dialled'].includes(s));
  console.log(`${unserved.length} country entries with a high risk class open ` +
              'outside the served set');
  if (findings.length === 0) return;
  console.warn(`  repair: POST ${VOICE}/DialingPermissions/BulkCountryUpdates ` +
               'with an UpdateRequest array disabling ' +
               'high_risk_special_numbers_enabled and ' +
               'high_risk_tollfraud_numbers_enabled for every unused ISO code');
  console.warn('  repair: run this on a schedule. Permissions get widened during ' +
               'incidents and the widening outlives the incident');
  if (unserved.length) 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_high_risk_dialing_audit import countries_for, money, prefix_index, verdict


def open_country(iso, low=True, special=False, fraud=False):
    return {"iso_code": iso, "country_codes": ["500"],
            "low_risk_numbers_enabled": low,
            "high_risk_special_numbers_enabled": special,
            "high_risk_tollfraud_numbers_enabled": fraud}


def test_both_classes_disabled_is_closed():
    assert verdict(open_country("LV"))[0] == "closed"


def test_low_risk_off_with_high_risk_on_is_the_telling_combination():
    state, detail = verdict(open_country("LV", low=False, fraud=True),
                            served=["US"])
    assert state == "premium-only"
    assert "Nobody configures that deliberately" in detail


def test_open_range_with_traffic_is_an_incident_not_an_exposure():
    state, detail = verdict(open_country("LV", special=True), served=["US"],
                            attempts=41, spend=1830.5)
    assert state == "open-and-dialled"
    assert "1830.50" in detail


def test_open_range_outside_the_served_set_is_carried_for_no_return():
    assert verdict(open_country("LV", fraud=True), served=["US", "GB"])[0] == \
        "open-unused"


def test_open_range_in_a_served_country_is_still_reported():
    state, _ = verdict(open_country("GB", special=True), served=["us", "gb"])
    assert state == "open-in-market"


def test_served_codes_are_compared_case_insensitively():
    assert verdict(open_country("GB", fraud=True), served=["gb"])[0] == "open-in-market"


def test_price_strings_are_negative_and_report_as_spend():
    assert money("-0.0850") == 0.085
    assert money(None) == 0.0
    assert money("not a price") == 0.0


def test_prefix_join_keeps_shared_codes_as_a_group():
    index = prefix_index([open_country("A"), open_country("B")])
    assert countries_for("+5005550100", index) == ["A", "B"]

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

const openCountry = (iso, low = true, special = false, fraud = false) => ({
  iso_code: iso,
  country_codes: ['500'],
  low_risk_numbers_enabled: low,
  high_risk_special_numbers_enabled: special,
  high_risk_tollfraud_numbers_enabled: fraud,
});

test('both classes disabled is closed', () => {
  assert.equal(verdict(openCountry('LV'))[0], 'closed');
});

test('low risk off with high risk on is the telling combination', () => {
  const [state, detail] = verdict(openCountry('LV', false, false, true), ['US']);
  assert.equal(state, 'premium-only');
  assert.match(detail, /Nobody configures that deliberately/);
});

test('open range with traffic is an incident not an exposure', () => {
  const [state, detail] = verdict(openCountry('LV', true, true), ['US'], 41, 1830.5);
  assert.equal(state, 'open-and-dialled');
  assert.match(detail, /1830\.50/);
});

test('open range outside the served set is carried for no return', () => {
  assert.equal(verdict(openCountry('LV', true, false, true), ['US', 'GB'])[0],
               'open-unused');
});

test('open range in a served country is still reported', () => {
  assert.equal(verdict(openCountry('GB', true, true), ['us', 'gb'])[0],
               'open-in-market');
});

test('price strings are negative and report as spend', () => {
  assert.equal(money('-0.0850'), 0.085);
  assert.equal(money(null), 0);
  assert.equal(money('not a price'), 0);
});

test('prefix join keeps shared codes as a group', () => {
  const index = prefixIndex([openCountry('A'), openCountry('B')]);
  assert.deepEqual(countriesFor('+5005550100', index), ['A', 'B']);
});