SMS Pumping Protection blocks legitimate OTPs with 30450

"""Report destinations blocked by Twilio SMS Pumping Protection (30450).

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 datetime as dt
import logging
import os
import sys
from email.utils import parsedate_to_datetime

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("twilio_pumping_block_audit")

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

# Both codes come out of the same fraud protection. Splitting them produces two
# reports about one event and no extra decision.
BLOCKED = (30450, 30485)

# Dialling codes, matched longest first. Without the length ordering every
# Bangladeshi number (880) lands in the North American bucket (1).
CODE_1 = {"1", "7"}
CODE_2 = {"20", "27", "30", "31", "32", "33", "34", "36", "39", "40", "41", "43",
          "44", "45", "46", "47", "48", "49", "51", "52", "53", "54", "55", "56",
          "57", "58", "60", "61", "62", "63", "64", "65", "66", "81", "82", "84",
          "86", "90", "91", "92", "93", "94", "95", "98"}
CODE_3 = {"211", "212", "213", "216", "218", "220", "221", "223", "225", "226",
          "227", "228", "229", "233", "234", "237", "243", "244", "249", "250",
          "251", "254", "255", "256", "260", "263", "264", "265", "267", "351",
          "352", "353", "354", "355", "356", "357", "358", "359", "370", "371",
          "372", "373", "374", "375", "376", "380", "381", "385", "386", "387",
          "389", "420", "421", "423", "500", "501", "502", "503", "504", "505",
          "506", "507", "508", "509", "852", "853", "855", "856", "880", "886",
          "960", "961", "962", "963", "964", "965", "966", "967", "968", "970",
          "971", "972", "973", "974", "975", "976", "977", "992", "993", "994",
          "995", "996", "998"}


def country_prefix(e164):
    """Dialling code for a destination number. Pure.

    Longest match wins, because the codes are a prefix-free set only when you
    read them that way: 880 has to be tested before 88 and before 1.
    """
    digits = "".join(c for c in str(e164 or "") if c.isdigit())
    if not digits:
        return "unknown"
    for size, table in ((3, CODE_3), (2, CODE_2), (1, CODE_1)):
        if digits[:size] in table:
            return digits[:size]
    return digits[:3]


def error_code(message):
    """Read error_code as an integer, or None.

    It is null on every healthy message and a number on failed ones, but some
    exports hand it back as a string. Comparing the raw value against 30450 is
    the mistake that reports a clean account in the middle of a block.
    """
    raw = message.get("error_code")
    if raw is None or raw == "":
        return None
    try:
        return int(raw)
    except (TypeError, ValueError):
        return None


def parse_ts(raw):
    """date_sent is RFC 2822 on this API. ISO is accepted too, because that is
    what fixtures and exports tend to carry."""
    s = str(raw or "").strip()
    if not s:
        return None
    stamp = None
    try:
        stamp = parsedate_to_datetime(s)
    except (TypeError, ValueError):
        try:
            stamp = dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
        except ValueError:
            return None
    if stamp is not None and stamp.tzinfo is None:
        stamp = stamp.replace(tzinfo=dt.timezone.utc)
    return stamp


def minutes_between(start, end):
    if start is None or end is None:
        return None
    return int((end - start).total_seconds() // 60)


def tally(messages, now):
    """Bucket outbound messages by destination dialling code. Pure, and `now`
    is an argument so the age of a block is testable without a clock.

    Inbound messages are skipped: they have no destination of ours and no
    delivery error worth counting.
    """
    rows = {}
    for m in messages:
        if str(m.get("direction") or "").startswith("inbound"):
            continue
        prefix = country_prefix(m.get("to"))
        row = rows.setdefault(prefix, {"total": 0, "blocked": 0, "sids": [],
                                       "first": None, "last": None})
        row["total"] += 1
        if error_code(m) in BLOCKED:
            row["blocked"] += 1
            if len(row["sids"]) < 3:
                row["sids"].append(m.get("sid"))
            stamp = parse_ts(m.get("date_sent") or m.get("date_created"))
            if stamp is not None:
                if row["first"] is None or stamp < row["first"]:
                    row["first"] = stamp
                if row["last"] is None or stamp > row["last"]:
                    row["last"] = stamp
    for row in rows.values():
        row["span_minutes"] = minutes_between(row["first"], row["last"])
        row["minutes_since_last"] = minutes_between(row["last"], now)
    return rows


def verdict(stats, min_blocked=3):
    """Classify one destination prefix. Pure, so the thresholds are visible
    rather than buried in a request loop.

    Returns (state, detail).
    """
    total = int(stats.get("total") or 0)
    blocked = int(stats.get("blocked") or 0)
    if not blocked:
        return ("clean", "%d message(s), none blocked" % total)

    rate = (blocked / total) if total else 1.0
    pct = rate * 100
    span = stats.get("span_minutes")
    since = stats.get("minutes_since_last")

    if blocked < min_blocked:
        return ("isolated",
                "%d of %d blocked (%.1f%%). Too few to separate a fraud block "
                "from an ordinary carrier reject, and Support wants at least %d "
                "Message SIDs before it will look."
                % (blocked, total, pct, min_blocked))

    if since is not None and since >= 60 and (span is None or span <= 240):
        return ("recovered",
                "%d of %d blocked (%.1f%%) inside a %s minute window that ended "
                "%d minutes ago. That is the shape of the temporary block: it "
                "lifted by itself, nobody was told, and the same prefix will hit "
                "it again." % (blocked, total, pct, span, since))

    if rate >= 0.5:
        return ("region-blocked",
                "%d of %d blocked (%.1f%%), last one %s minutes ago. More than "
                "half of everything to this prefix is being refused: treat it as "
                "an outage for that country, not as noise."
                % (blocked, total, pct, since))

    return ("intermittent",
            "%d of %d blocked (%.1f%%) spread over %s minutes. Recurring rather "
            "than one burst, so a safe list entry is worth more than waiting it "
            "out." % (blocked, total, pct, span))


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_messages(session, account, since, limit):
    """Page Messages.json. There is no Status or ErrorCode filter on this
    resource, so the date window and the page cap are the only bounds there
    are."""
    url = "%s/Accounts/%s/Messages.json" % (BASE, account)
    params = {"PageSize": 1000, "DateSent>=": since}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("messages", []))
        nxt = page.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("--days", type=int, default=3,
                    help="how far back to read the Messages list")
    ap.add_argument("--max-messages", type=int, default=20000,
                    help="stop paging after this many messages")
    ap.add_argument("--min-blocked", type=int, default=3,
                    help="fewer than this on one prefix is reported as isolated")
    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=args.days)).isoformat()
    messages = list_messages(session, account, since, args.max_messages)
    if not messages:
        log.info("no messages sent since %s", since)
        return 0

    now = dt.datetime.now(dt.timezone.utc)
    prefixes = tally(messages, now)
    bad = 0
    for prefix, stats in sorted(prefixes.items()):
        state, detail = verdict(stats, args.min_blocked)
        line = "%-15s +%-5s %s" % (state, prefix, detail)
        if state == "clean":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        log.warning("  message sids: %s", ", ".join(str(s) for s in stats["sids"]))
        log.warning("  repair: no API call lifts a 30450. Add the verified "
                    "numbers or the +%s prefix to the Global Safe List (Console "
                    "-> Messaging -> Settings -> Global Safe List), or send that "
                    "route with RiskCheck=disable. Keep RiskCheck on elsewhere.",
                    prefix)

    log.info("%d destination prefix(es) over %d day(s), %d blocked",
             len(prefixes), args.days, bad)
    return 1 if bad else 0


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

/**
 * Report destinations blocked by Twilio SMS Pumping Protection (30450).
 *
 * 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`;

// Both codes come out of the same fraud protection. Splitting them produces two
// reports about one event and no extra decision.
const BLOCKED = new Set([30450, 30485]);

// Dialling codes, matched longest first. Without the length ordering every
// Bangladeshi number (880) lands in the North American bucket (1).
const CODE_1 = new Set(['1', '7']);
const CODE_2 = new Set(['20', '27', '30', '31', '32', '33', '34', '36', '39', '40',
  '41', '43', '44', '45', '46', '47', '48', '49', '51', '52', '53', '54', '55',
  '56', '57', '58', '60', '61', '62', '63', '64', '65', '66', '81', '82', '84',
  '86', '90', '91', '92', '93', '94', '95', '98']);
const CODE_3 = new Set(['211', '212', '213', '216', '218', '220', '221', '223',
  '225', '226', '227', '228', '229', '233', '234', '237', '243', '244', '249',
  '250', '251', '254', '255', '256', '260', '263', '264', '265', '267', '351',
  '352', '353', '354', '355', '356', '357', '358', '359', '370', '371', '372',
  '373', '374', '375', '376', '380', '381', '385', '386', '387', '389', '420',
  '421', '423', '500', '501', '502', '503', '504', '505', '506', '507', '508',
  '509', '852', '853', '855', '856', '880', '886', '960', '961', '962', '963',
  '964', '965', '966', '967', '968', '970', '971', '972', '973', '974', '975',
  '976', '977', '992', '993', '994', '995', '996', '998']);

/**
 * Dialling code for a destination number. Pure. Longest match wins, because the
 * codes are a prefix-free set only when you read them that way.
 */
export function countryPrefix(e164) {
  const digits = String(e164 ?? '').replace(/\D/g, '');
  if (!digits) return 'unknown';
  for (const [size, table] of [[3, CODE_3], [2, CODE_2], [1, CODE_1]]) {
    if (table.has(digits.slice(0, size))) return digits.slice(0, size);
  }
  return digits.slice(0, 3);
}

/**
 * Read error_code as a number, or null. It is null on healthy messages and a
 * number on failed ones; comparing the raw value is how the audit reports a
 * clean account in the middle of a block.
 */
export function errorCode(message) {
  const raw = message.error_code;
  if (raw === null || raw === undefined || raw === '') return null;
  const n = Number(raw);
  return Number.isFinite(n) ? n : null;
}

/** date_sent is RFC 2822 on this API; ISO is accepted too. */
export function parseTs(raw) {
  const s = String(raw ?? '').trim();
  if (!s) return null;
  const t = Date.parse(s);
  return Number.isNaN(t) ? null : new Date(t);
}

function minutesBetween(start, end) {
  if (!start || !end) return null;
  return Math.floor((end.getTime() - start.getTime()) / 60000);
}

/**
 * Bucket outbound messages by destination dialling code. Pure, and `now` is an
 * argument so the age of a block is testable without a clock.
 */
export function tally(messages, now) {
  const rows = new Map();
  for (const m of messages) {
    if (String(m.direction ?? '').startsWith('inbound')) continue;
    const prefix = countryPrefix(m.to);
    if (!rows.has(prefix)) {
      rows.set(prefix, { total: 0, blocked: 0, sids: [], first: null, last: null });
    }
    const row = rows.get(prefix);
    row.total += 1;
    if (BLOCKED.has(errorCode(m))) {
      row.blocked += 1;
      if (row.sids.length < 3) row.sids.push(m.sid);
      const stamp = parseTs(m.date_sent ?? m.date_created);
      if (stamp) {
        if (!row.first || stamp < row.first) row.first = stamp;
        if (!row.last || stamp > row.last) row.last = stamp;
      }
    }
  }
  for (const row of rows.values()) {
    row.span_minutes = minutesBetween(row.first, row.last);
    row.minutes_since_last = minutesBetween(row.last, now);
  }
  return rows;
}

/**
 * Classify one destination prefix. Pure, so the thresholds are visible rather
 * than buried in a request loop. Returns [state, detail].
 */
export function verdict(stats, minBlocked = 3) {
  const total = Number(stats.total ?? 0);
  const blocked = Number(stats.blocked ?? 0);
  if (!blocked) return ['clean', `${total} message(s), none blocked`];

  const rate = total ? blocked / total : 1;
  const pct = (rate * 100).toFixed(1);
  const span = stats.span_minutes;
  const since = stats.minutes_since_last;

  if (blocked < minBlocked) {
    return ['isolated',
      `${blocked} of ${total} blocked (${pct}%). Too few to separate a fraud ` +
      'block from an ordinary carrier reject, and Support wants at least ' +
      `${minBlocked} Message SIDs before it will look.`];
  }

  if (since !== null && since !== undefined && since >= 60 &&
      (span === null || span === undefined || span <= 240)) {
    return ['recovered',
      `${blocked} of ${total} blocked (${pct}%) inside a ${span} minute window ` +
      `that ended ${since} minutes ago. That is the shape of the temporary ` +
      'block: it lifted by itself, nobody was told, and the same prefix will ' +
      'hit it again.'];
  }

  if (rate >= 0.5) {
    return ['region-blocked',
      `${blocked} of ${total} blocked (${pct}%), last one ${since} minutes ago. ` +
      'More than half of everything to this prefix is being refused: treat it ' +
      'as an outage for that country, not as noise.'];
  }

  return ['intermittent',
    `${blocked} of ${total} blocked (${pct}%) spread over ${span} minutes. ` +
    'Recurring rather than one burst, so a safe list entry is worth more than ' +
    'waiting it out.'];
}

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 listMessages(auth, account, since, limit) {
  let url = `${BASE}/Accounts/${account}/Messages.json`;
  let params = { PageSize: 1000, 'DateSent>=': since };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.messages ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

function flag(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 = flag('--days', 3);
  const minBlocked = flag('--min-blocked', 3);

  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);
  const messages = await listMessages(auth, account, since, flag('--max-messages', 20000));
  if (messages.length === 0) {
    console.log(`no messages sent since ${since}`);
    return;
  }

  const prefixes = tally(messages, new Date());
  let bad = 0;
  for (const prefix of [...prefixes.keys()].sort()) {
    const stats = prefixes.get(prefix);
    const [state, detail] = verdict(stats, minBlocked);
    const line = `${state.padEnd(15)} +${prefix.padEnd(5)} ${detail}`;
    if (state === 'clean') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    console.warn(`  message sids: ${stats.sids.join(', ')}`);
    console.warn('  repair: no API call lifts a 30450. Add the verified numbers ' +
                 `or the +${prefix} prefix to the Global Safe List (Console -> ` +
                 'Messaging -> Settings -> Global Safe List), or send that route ' +
                 'with RiskCheck=disable. Keep RiskCheck on elsewhere.');
  }

  console.log(`${prefixes.size} destination prefix(es) over ${days} day(s), ${bad} blocked`);
  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 file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

import datetime as dt

from twilio_pumping_block_audit import country_prefix, tally, verdict

NOW = dt.datetime(2026, 3, 2, 12, 0, tzinfo=dt.timezone.utc)


def blocked(sid, to, sent):
    return {"sid": sid, "to": to, "error_code": 30450, "status": "failed",
            "date_sent": sent}


def test_dialling_codes_match_longest_first():
    assert country_prefix("+8801711000000") == "880"
    assert country_prefix("+447700900000") == "44"
    assert country_prefix("+15551230000") == "1"


def test_prefix_of_junk_is_not_a_crash():
    assert country_prefix(None) == "unknown"
    assert country_prefix("not a number") == "unknown"


def test_error_code_as_a_string_still_counts():
    rows = tally([{"sid": "SM1", "to": "+8801711000000", "error_code": "30450",
                   "date_sent": "Mon, 02 Mar 2026 09:00:00 +0000"}], NOW)
    assert rows["880"]["blocked"] == 1


def test_tally_groups_by_prefix_and_skips_inbound():
    rows = tally([
        blocked("SM1", "+8801711000000", "Mon, 02 Mar 2026 09:00:00 +0000"),
        blocked("SM2", "+8801711000001", "Mon, 02 Mar 2026 09:11:00 +0000"),
        {"sid": "SM3", "to": "+15551230000", "status": "delivered"},
        {"sid": "SM4", "to": "+15551230000", "direction": "inbound"},
    ], NOW)
    assert sorted(rows) == ["1", "880"]
    assert rows["880"]["blocked"] == 2
    assert rows["880"]["span_minutes"] == 11
    assert rows["880"]["minutes_since_last"] == 169
    assert rows["1"]["total"] == 1


def test_a_burst_that_already_stopped_reads_as_recovered():
    state, detail = verdict({"total": 400, "blocked": 94, "span_minutes": 11,
                             "minutes_since_last": 169})
    assert state == "recovered"
    assert "lifted by itself" in detail


def test_a_prefix_still_failing_now_is_an_outage_not_a_blip():
    state, detail = verdict({"total": 10, "blocked": 8, "span_minutes": 600,
                             "minutes_since_last": 4})
    assert state == "region-blocked"
    assert "outage" in detail


def test_recurring_low_rate_is_intermittent():
    state, _ = verdict({"total": 500, "blocked": 40, "span_minutes": 3000,
                        "minutes_since_last": 6})
    assert state == "intermittent"


def test_two_blocked_is_too_few_to_escalate():
    state, detail = verdict({"total": 50, "blocked": 2, "span_minutes": 3,
                             "minutes_since_last": 400})
    assert state == "isolated"
    assert "at least 3" in detail


def test_no_blocked_messages_is_clean():
    state, detail = verdict({"total": 900, "blocked": 0})
    assert state == "clean"
    assert "900" in detail


def test_sids_are_capped_at_the_three_support_asks_for():
    rows = tally([blocked("SM%d" % i, "+8801711000000",
                          "Mon, 02 Mar 2026 09:00:00 +0000") for i in range(9)], NOW)
    assert rows["880"]["sids"] == ["SM0", "SM1", "SM2"]
    assert rows["880"]["blocked"] == 9

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { countryPrefix, tally, verdict } from './twilio-pumping-block-audit.mjs';

const NOW = new Date('2026-03-02T12:00:00Z');

const blocked = (sid, to, sent) => ({
  sid, to, error_code: 30450, status: 'failed', date_sent: sent,
});

test('dialling codes match longest first', () => {
  assert.equal(countryPrefix('+8801711000000'), '880');
  assert.equal(countryPrefix('+447700900000'), '44');
  assert.equal(countryPrefix('+15551230000'), '1');
});

test('prefix of junk is not a crash', () => {
  assert.equal(countryPrefix(null), 'unknown');
  assert.equal(countryPrefix('not a number'), 'unknown');
});

test('error_code as a string still counts', () => {
  const rows = tally([{ sid: 'SM1', to: '+8801711000000', error_code: '30450',
                        date_sent: 'Mon, 02 Mar 2026 09:00:00 +0000' }], NOW);
  assert.equal(rows.get('880').blocked, 1);
});

test('tally groups by prefix and skips inbound', () => {
  const rows = tally([
    blocked('SM1', '+8801711000000', 'Mon, 02 Mar 2026 09:00:00 +0000'),
    blocked('SM2', '+8801711000001', 'Mon, 02 Mar 2026 09:11:00 +0000'),
    { sid: 'SM3', to: '+15551230000', status: 'delivered' },
    { sid: 'SM4', to: '+15551230000', direction: 'inbound' },
  ], NOW);
  assert.deepEqual([...rows.keys()].sort(), ['1', '880']);
  assert.equal(rows.get('880').blocked, 2);
  assert.equal(rows.get('880').span_minutes, 11);
  assert.equal(rows.get('880').minutes_since_last, 169);
  assert.equal(rows.get('1').total, 1);
});

test('a burst that already stopped reads as recovered', () => {
  const [state, detail] = verdict({ total: 400, blocked: 94, span_minutes: 11,
                                    minutes_since_last: 169 });
  assert.equal(state, 'recovered');
  assert.match(detail, /lifted by itself/);
});

test('a prefix still failing now is an outage, not a blip', () => {
  const [state, detail] = verdict({ total: 10, blocked: 8, span_minutes: 600,
                                    minutes_since_last: 4 });
  assert.equal(state, 'region-blocked');
  assert.match(detail, /outage/);
});

test('recurring low rate is intermittent', () => {
  assert.equal(verdict({ total: 500, blocked: 40, span_minutes: 3000,
                         minutes_since_last: 6 })[0], 'intermittent');
});

test('two blocked is too few to escalate', () => {
  const [state, detail] = verdict({ total: 50, blocked: 2, span_minutes: 3,
                                    minutes_since_last: 400 });
  assert.equal(state, 'isolated');
  assert.match(detail, /at least 3/);
});

test('no blocked messages is clean', () => {
  const [state, detail] = verdict({ total: 900, blocked: 0 });
  assert.equal(state, 'clean');
  assert.match(detail, /900/);
});

test('sids are capped at the three Support asks for', () => {
  const rows = tally([...Array(9).keys()].map((i) =>
    blocked(`SM${i}`, '+8801711000000', 'Mon, 02 Mar 2026 09:00:00 +0000')), NOW);
  assert.deepEqual(rows.get('880').sids, ['SM0', 'SM1', 'SM2']);
  assert.equal(rows.get('880').blocked, 9);
});