Fraud Guard blocked the prefix, so real users get 60410

"""Find number prefixes Fraud Guard has blocked, which fail real users with 60410.

Fraud Guard blocks SMS to a prefix for twelve hours when it sees pumping-shaped
traffic, and re-arms while the pattern continues. There is no unblock API: the
block ends when the traffic causing it stops.

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.
"""
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_fraud_guard_block_audit")

VERIFY = "https://verify.twilio.com/v2"
LOOKUPS = "https://lookups.twilio.com/v2"

# Twilio's own guidance for gating a signup on the score: block at 90 and above,
# add friction in the middle band.
BLOCK_SCORE = 90
FRICTION_SCORE = 60

# Below this many unconverted attempts a prefix is not a cluster, it is a few
# people whose phones were off.
MIN_ATTEMPTS = 5


def prefix_of(number, digits=6):
    """Leading digits of an E.164 number. Fraud Guard acts on ranges, so the
    prefix is the unit of both the block and the report.
    """
    n = "".join(c for c in str(number or "") if c.isdigit())
    return ("+" + n[:digits]) if n else "?"


def group_attempts(attempts, digits=6):
    """Bucket unconverted attempts by (country, prefix). Pure, so the grouping
    can be tested without a network.
    """
    groups = {}
    for a in attempts:
        to = (a.get("channel_data") or {}).get("to")
        keyed = (a.get("country") or "??", prefix_of(to, digits))
        g = groups.setdefault(keyed, {"country": keyed[0], "prefix": keyed[1],
                                      "attempts": 0, "sample": None})
        g["attempts"] += 1
        if g["sample"] is None and to:
            g["sample"] = to
    return sorted(groups.values(), key=lambda g: -g["attempts"])


def verdict(group, risk, min_attempts=MIN_ATTEMPTS):
    """Classify one (country, prefix) group against Lookup's pumping risk.

    `risk` is the sms_pumping_risk object from Lookup, or None when the field was
    not returned. Pure, so the five states can be tested without a network.

    Returns (state, detail).
    """
    attempts = int(group.get("attempts") or 0)
    where = "%s %s" % (group.get("country", "??"), group.get("prefix", "?"))

    if attempts < min_attempts:
        return ("thin",
                "%s: %d unconverted attempt(s), below the %d cluster floor"
                % (where, attempts, min_attempts))

    if not risk:
        return ("no-risk-data",
                "%s: %d unconverted, and Lookup returned no sms_pumping_risk. "
                "That field is billed and entitlement-gated: confirm the add-on "
                "before reading this as clear." % (where, attempts))

    score = risk.get("sms_pumping_risk_score")
    score_txt = "score %s" % ("?" if score is None else score)
    carrier = risk.get("carrier_risk_category") or "unknown"

    if risk.get("number_blocked"):
        return ("blocked",
                "%s: Fraud Guard block is live (since %s, %s, carrier risk %s) "
                "on %d unconverted attempts. Every real user on this prefix gets "
                "60410 for twelve hours, and it re-arms while the traffic "
                "continues. There is no unblock API."
                % (where, risk.get("number_blocked_date") or "unknown date",
                   score_txt, carrier, attempts))

    recent = int(risk.get("number_blocked_last_3_months") or 0)
    if recent > 0:
        return ("blocked-recently",
                "%s: not blocked now, but blocked %d time(s) in three months "
                "(%s, carrier risk %s). The source traffic is still arriving, so "
                "this range will block again." % (where, recent, score_txt, carrier))

    if score is not None and score >= BLOCK_SCORE:
        return ("high-risk",
                "%s: %s on %d unconverted attempts. This is the traffic Fraud "
                "Guard blocks; gate signup on the score before it does."
                % (where, score_txt, attempts))

    if score is not None and score >= FRICTION_SCORE:
        return ("watch",
                "%s: %s, in the band where friction belongs rather than a hard "
                "block (carrier risk %s)." % (where, score_txt, carrier))

    return ("clear",
            "%s: %s, no block on record. The %d unconverted attempts here are "
            "something else." % (where, score_txt, attempts))


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 unconverted(session, service, since, limit=2000):
    url = VERIFY + "/Attempts"
    params = {"VerifyServiceSid": service, "Status": "unconverted",
              "DateCreatedAfter": since, "PageSize": 100}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("attempts", []))
        url, params = (page.get("meta") or {}).get("next_page_url"), {}
    return out[:limit]


def pumping_risk(session, e164):
    """One billed Lookup per prefix group, not per number."""
    body = get(session, "%s/PhoneNumbers/%s" % (LOOKUPS, e164),
               Fields="sms_pumping_risk")
    return body.get("sms_pumping_risk")


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--service", required=True, help="Verify Service SID (VA...)")
    ap.add_argument("--days", type=int, default=2, help="window to sweep")
    ap.add_argument("--prefix-digits", type=int, default=6,
                    help="leading digits that define a range")
    ap.add_argument("--max-lookups", type=int, default=20,
                    help="cap on billed Lookup calls")
    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.datetime.now(dt.timezone.utc)
             - dt.timedelta(days=args.days)).strftime("%Y-%m-%dT%H:%M:%SZ")

    groups = group_attempts(unconverted(session, args.service, since),
                            args.prefix_digits)
    if not groups:
        log.info("no unconverted attempts in the last %d day(s)", args.days)
        return 0

    blocked = 0
    for i, g in enumerate(groups):
        risk = None
        if g["sample"] and i < args.max_lookups and g["attempts"] >= MIN_ATTEMPTS:
            risk = pumping_risk(session, g["sample"])
        state, detail = verdict(g, risk)
        line = "%-16s %s" % (state, detail)
        if state in ("blocked", "blocked-recently", "high-risk"):
            blocked += state == "blocked"
            log.warning(line)
            log.warning("  repair: no API lifts this. Add an IP-keyed Service "
                        "Rate Limit on %s, gate signup on "
                        "sms_pumping_risk_score (block at %d, friction from %d), "
                        "and lower the level at Console > Verify > Services > "
                        "SMS if this is a false positive on your own traffic",
                        args.service, BLOCK_SCORE, FRICTION_SCORE)
        else:
            log.info(line)

    log.info("%d prefix group(s), %d currently blocked", len(groups), blocked)
    return 1 if blocked else 0


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

/**
 * Find number prefixes Fraud Guard has blocked, which fail real users with 60410.
 *
 * Fraud Guard blocks SMS to a prefix for twelve hours when it sees pumping-shaped
 * traffic, and re-arms while the pattern continues. There is no unblock API: the
 * block ends when the traffic causing it stops.
 *
 * 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 VERIFY = 'https://verify.twilio.com/v2';
const LOOKUPS = 'https://lookups.twilio.com/v2';

// Twilio's own guidance for gating a signup on the score.
export const BLOCK_SCORE = 90;
export const FRICTION_SCORE = 60;

// Below this many unconverted attempts a prefix is not a cluster.
export const MIN_ATTEMPTS = 5;

/** Leading digits of an E.164 number: Fraud Guard acts on ranges. */
export function prefixOf(number, digits = 6) {
  const n = String(number ?? '').replace(/\D/g, '');
  return n ? `+${n.slice(0, digits)}` : '?';
}

/**
 * Bucket unconverted attempts by (country, prefix). Pure, so the grouping can be
 * tested without a network.
 */
export function groupAttempts(attempts, digits = 6) {
  const groups = new Map();
  for (const a of attempts) {
    const to = a.channel_data?.to;
    const country = a.country ?? '??';
    const prefix = prefixOf(to, digits);
    const key = `${country} ${prefix}`;
    if (!groups.has(key)) {
      groups.set(key, { country, prefix, attempts: 0, sample: null });
    }
    const g = groups.get(key);
    g.attempts += 1;
    if (g.sample === null && to) g.sample = to;
  }
  return [...groups.values()].sort((x, y) => y.attempts - x.attempts);
}

/**
 * Classify one (country, prefix) group against Lookup's pumping risk. `risk` is
 * the sms_pumping_risk object, or null when the field was not returned. Pure, so
 * the five states can be tested without a network. Returns [state, detail].
 */
export function verdict(group, risk, minAttempts = MIN_ATTEMPTS) {
  const attempts = Number(group.attempts ?? 0);
  const where = `${group.country ?? '??'} ${group.prefix ?? '?'}`;

  if (attempts < minAttempts) {
    return ['thin',
      `${where}: ${attempts} unconverted attempt(s), below the ${minAttempts} ` +
      'cluster floor'];
  }

  if (!risk) {
    return ['no-risk-data',
      `${where}: ${attempts} unconverted, and Lookup returned no ` +
      'sms_pumping_risk. That field is billed and entitlement-gated: confirm ' +
      'the add-on before reading this as clear.'];
  }

  const score = risk.sms_pumping_risk_score;
  const scoreTxt = `score ${score ?? '?'}`;
  const carrier = risk.carrier_risk_category ?? 'unknown';

  if (risk.number_blocked) {
    return ['blocked',
      `${where}: Fraud Guard block is live (since ` +
      `${risk.number_blocked_date ?? 'unknown date'}, ${scoreTxt}, carrier ` +
      `risk ${carrier}) on ${attempts} unconverted attempts. Every real user ` +
      'on this prefix gets 60410 for twelve hours, and it re-arms while the ' +
      'traffic continues. There is no unblock API.'];
  }

  const recent = Number(risk.number_blocked_last_3_months ?? 0);
  if (recent > 0) {
    return ['blocked-recently',
      `${where}: not blocked now, but blocked ${recent} time(s) in three ` +
      `months (${scoreTxt}, carrier risk ${carrier}). The source traffic is ` +
      'still arriving, so this range will block again.'];
  }

  if (score !== undefined && score !== null && score >= BLOCK_SCORE) {
    return ['high-risk',
      `${where}: ${scoreTxt} on ${attempts} unconverted attempts. This is the ` +
      'traffic Fraud Guard blocks; gate signup on the score before it does.'];
  }

  if (score !== undefined && score !== null && score >= FRICTION_SCORE) {
    return ['watch',
      `${where}: ${scoreTxt}, in the band where friction belongs rather than a ` +
      `hard block (carrier risk ${carrier}).`];
  }

  return ['clear',
    `${where}: ${scoreTxt}, no block on record. The ${attempts} unconverted ` +
    'attempts here are something else.'];
}

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 unconverted(auth, service, since, limit = 2000) {
  let url = `${VERIFY}/Attempts`;
  let params = { VerifyServiceSid: service, Status: 'unconverted',
                 DateCreatedAfter: since, PageSize: 100 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.attempts ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

/** One billed Lookup per prefix group, not per number. */
async function pumpingRisk(auth, e164) {
  const body = await get(auth, `${LOOKUPS}/PhoneNumbers/${e164}`,
                         { Fields: 'sms_pumping_risk' });
  return body.sms_pumping_risk ?? null;
}

function arg(name, fallback) {
  const i = process.argv.indexOf(name);
  return i === -1 ? fallback : 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 service = arg('--service');
  if (!service) {
    console.error('pass --service VA...');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const days = Number(arg('--days', '2'));
  const maxLookups = Number(arg('--max-lookups', '20'));
  const since = new Date(Date.now() - days * 86400000).toISOString()
    .replace(/\.\d+Z$/, 'Z');

  const groups = groupAttempts(await unconverted(auth, service, since));
  if (groups.length === 0) {
    console.log(`no unconverted attempts in the last ${days} day(s)`);
    return;
  }

  let blocked = 0;
  for (const [i, g] of groups.entries()) {
    let risk = null;
    if (g.sample && i < maxLookups && g.attempts >= MIN_ATTEMPTS) {
      risk = await pumpingRisk(auth, g.sample);
    }
    const [state, detail] = verdict(g, risk);
    const line = `${state.padEnd(16)} ${detail}`;
    if (state === 'blocked' || state === 'blocked-recently' || state === 'high-risk') {
      if (state === 'blocked') blocked += 1;
      console.warn(line);
      console.warn('  repair: no API lifts this. Add an IP-keyed Service Rate ' +
                   `Limit on ${service}, gate signup on sms_pumping_risk_score ` +
                   `(block at ${BLOCK_SCORE}, friction from ${FRICTION_SCORE}), ` +
                   'and lower the level at Console > Verify > Services > SMS if ' +
                   'this is a false positive on your own traffic');
    } else {
      console.log(line);
    }
  }

  console.log(`${groups.length} prefix group(s), ${blocked} currently blocked`);
  process.exitCode = blocked ? 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_fraud_guard_block_audit import group_attempts, prefix_of, verdict

GROUP = {"country": "GB", "prefix": "+447700", "attempts": 44, "sample": "+447700900123"}


def test_live_block_is_the_incident():
    state, detail = verdict(GROUP, {"number_blocked": True,
                                    "number_blocked_date": "2026-08-29",
                                    "sms_pumping_risk_score": 97,
                                    "carrier_risk_category": "high"})
    assert state == "blocked"
    assert "60410" in detail
    assert "no unblock API" in detail


def test_blocked_before_but_not_now_is_a_source_problem():
    state, detail = verdict(GROUP, {"number_blocked": False,
                                    "number_blocked_last_3_months": 2,
                                    "sms_pumping_risk_score": 71})
    assert state == "blocked-recently"
    assert "block again" in detail


def test_high_score_with_no_block_yet_is_its_own_state():
    state, detail = verdict(GROUP, {"number_blocked": False,
                                    "number_blocked_last_3_months": 0,
                                    "sms_pumping_risk_score": 94})
    assert state == "high-risk"
    assert "before it does" in detail


def test_middle_band_asks_for_friction_not_a_block():
    state, _ = verdict(GROUP, {"number_blocked": False,
                               "number_blocked_last_3_months": 0,
                               "sms_pumping_risk_score": 66})
    assert state == "watch"


def test_missing_pumping_risk_is_never_reported_as_clear():
    # An unentitled field and a clean number look the same in the response.
    state, detail = verdict(GROUP, None)
    assert state == "no-risk-data"
    assert "entitlement-gated" in detail


def test_a_handful_of_attempts_is_not_a_cluster():
    state, _ = verdict({"country": "GB", "prefix": "+447700", "attempts": 2},
                       {"number_blocked": True})
    assert state == "thin"


def test_attempts_group_by_country_and_prefix_keeping_a_sample():
    groups = group_attempts([
        {"country": "GB", "channel_data": {"to": "+447700900123"}},
        {"country": "GB", "channel_data": {"to": "+447700900456"}},
        {"country": "FR", "channel_data": {"to": "+33612345678"}},
    ])
    assert groups[0]["prefix"] == "+447700"
    assert groups[0]["attempts"] == 2
    assert groups[0]["sample"] == "+447700900123"
    assert prefix_of("+33 6 12 34 56 78") == "+336123"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { groupAttempts, prefixOf, verdict } from './twilio-fraud-guard-block-audit.mjs';

const GROUP = { country: 'GB', prefix: '+447700', attempts: 44, sample: '+447700900123' };

test('live block is the incident', () => {
  const [state, detail] = verdict(GROUP, {
    number_blocked: true, number_blocked_date: '2026-08-29',
    sms_pumping_risk_score: 97, carrier_risk_category: 'high' });
  assert.equal(state, 'blocked');
  assert.match(detail, /60410/);
  assert.match(detail, /no unblock API/);
});

test('blocked before but not now is a source problem', () => {
  const [state, detail] = verdict(GROUP, {
    number_blocked: false, number_blocked_last_3_months: 2,
    sms_pumping_risk_score: 71 });
  assert.equal(state, 'blocked-recently');
  assert.match(detail, /block again/);
});

test('high score with no block yet is its own state', () => {
  const [state, detail] = verdict(GROUP, {
    number_blocked: false, number_blocked_last_3_months: 0,
    sms_pumping_risk_score: 94 });
  assert.equal(state, 'high-risk');
  assert.match(detail, /before it does/);
});

test('middle band asks for friction not a block', () => {
  assert.equal(verdict(GROUP, {
    number_blocked: false, number_blocked_last_3_months: 0,
    sms_pumping_risk_score: 66 })[0], 'watch');
});

test('missing pumping risk is never reported as clear', () => {
  const [state, detail] = verdict(GROUP, null);
  assert.equal(state, 'no-risk-data');
  assert.match(detail, /entitlement-gated/);
});

test('a handful of attempts is not a cluster', () => {
  assert.equal(
    verdict({ country: 'GB', prefix: '+447700', attempts: 2 },
            { number_blocked: true })[0], 'thin');
});

test('attempts group by country and prefix keeping a sample', () => {
  const groups = groupAttempts([
    { country: 'GB', channel_data: { to: '+447700900123' } },
    { country: 'GB', channel_data: { to: '+447700900456' } },
    { country: 'FR', channel_data: { to: '+33612345678' } },
  ]);
  assert.equal(groups[0].prefix, '+447700');
  assert.equal(groups[0].attempts, 2);
  assert.equal(groups[0].sample, '+447700900123');
  assert.equal(prefixOf('+33 6 12 34 56 78'), '+336123');
});