Verify conversion collapses in one country: SMS pumping

"""Find Verify countries whose conversion rate has collapsed against the baseline.

A collapse in one country on rising volume is SMS pumping in progress: the OTP
is delivered and billed, and nobody was ever going to enter it.

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 start billable
verifications.
"""
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_verify_conversion_audit")

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

# A country is only judged once it has this many attempts in the window. Three
# attempts and one conversion is 33%, and it means nothing at all.
MIN_ATTEMPTS = 40

# Fractions of the service baseline, not absolute rates. A service doing web
# signups converts near 70%; one attached to a re-engagement flow near 25%. Any
# fixed threshold is wrong for one of them.
COLLAPSE_RATIO = 0.35
WATCH_RATIO = 0.70


def conversion_rate(row):
    """Conversion rate for one summary row, as a percentage or None.

    Prefers conversion_rate_percentage as returned, and falls back to the counts
    so the function still works on a row assembled from total_converted and
    total_attempts alone.
    """
    pct = row.get("conversion_rate_percentage")
    if pct is not None:
        return float(pct)
    total = int(row.get("total_attempts") or 0)
    if total <= 0:
        return None
    return 100.0 * float(row.get("total_converted") or 0) / total


def verdict(row, baseline, min_attempts=MIN_ATTEMPTS):
    """Classify one country's summary against the service baseline.

    Pure, so the two rules that matter -- relative to baseline, and only above a
    volume floor -- can be tested without a network.

    Returns (state, detail).
    """
    attempts = int(row.get("total_attempts") or 0)
    country = row.get("country") or "??"

    if attempts <= 0:
        return ("no-traffic", "no attempts in the window")

    if baseline is None or baseline <= 0:
        return ("no-baseline",
                "the service baseline is zero or missing, so nothing can be "
                "compared against it: widen the window before reading this run")

    rate = conversion_rate(row)
    if rate is None:
        return ("no-baseline", "no conversion rate on the row")

    ratio = rate / baseline
    shape = ("%s: %.1f%% conversion against a %.1f%% baseline on %d attempts"
             % (country, rate, baseline, attempts))

    if attempts < min_attempts:
        return ("thin",
                "%s, below the %d attempt floor: too few to read as anything"
                % (shape, min_attempts))

    if ratio <= COLLAPSE_RATIO:
        return ("collapse",
                "%s (%.0f%% of baseline). The sends succeeded and were billed, "
                "and nobody entered the code: that is the shape of SMS pumping, "
                "not a broken integration." % (shape, ratio * 100))

    if ratio <= WATCH_RATIO:
        return ("watch",
                "%s (%.0f%% of baseline). Below the service, not yet at collapse: "
                "worth a second window before acting." % (shape, ratio * 100))

    return ("healthy", "%s (%.0f%% of baseline)" % (shape, ratio * 100))


def prefix_of(number, digits=6):
    """Leading digits of an E.164 number. Pumping concentrates on a few ranges
    inside a country, and the prefix is what the repair is written against.
    """
    n = "".join(c for c in str(number or "") if c.isdigit())
    return ("+" + n[:digits]) if n else "?"


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 summary(session, service, since, country=None):
    params = {"VerifyServiceSid": service, "DateCreatedAfter": since}
    if country:
        params["Country"] = country
    return get(session, VERIFY + "/Attempts/Summary", **params)


def unconverted(session, service, since, limit=1000):
    """One bounded sweep of unconverted attempts, used for two things: which
    countries are worth asking the summary about, and which prefixes inside them
    are carrying the traffic.
    """
    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", []))
        nxt = (page.get("meta") or {}).get("next_page_url")
        url, params = nxt, {}
    return out[:limit]


def countries_seen(attempts):
    """Countries in the unconverted sweep, most recent traffic first."""
    seen = {}
    for a in attempts:
        code = a.get("country")
        if not code:
            continue
        seen[code] = seen.get(code, 0) + 1
    return [c for c, _ in sorted(seen.items(), key=lambda kv: -kv[1])]


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=7, help="window to summarise")
    ap.add_argument("--country", action="append", default=[],
                    help="ISO 3166-1 alpha-2 code; repeatable. Default: the "
                         "countries seen in unconverted attempts")
    ap.add_argument("--min-attempts", type=int, default=MIN_ATTEMPTS,
                    help="volume floor below which a rate is not read")
    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")

    base_row = summary(session, args.service, since)
    baseline = conversion_rate(base_row)
    log.info("baseline %.1f%% over %s attempts",
             baseline or 0.0, base_row.get("total_attempts", 0))

    attempts = unconverted(session, args.service, since)
    countries = args.country or countries_seen(attempts)
    if not countries:
        log.info("no countries to check in the last %d day(s)", args.days)
        return 0

    bad = 0
    for code in countries:
        row = summary(session, args.service, since, country=code)
        row.setdefault("country", code)
        state, detail = verdict(row, baseline, args.min_attempts)
        line = "%-10s %s" % (state, detail)
        if state in ("collapse", "watch"):
            bad += state == "collapse"
            log.warning(line)
            hot = {}
            for a in attempts:
                if a.get("country") == code:
                    p = prefix_of((a.get("channel_data") or {}).get("to"))
                    hot[p] = hot.get(p, 0) + 1
            for p, n in sorted(hot.items(), key=lambda kv: -kv[1])[:3]:
                log.warning("  %s x%d unconverted", p, n)
            log.warning("  repair: Console > Verify > Services > %s > SMS: "
                        "enable Fraud Guard, restrict Geo Permissions to the "
                        "countries you serve, and add an IP-keyed Service Rate "
                        "Limit", args.service)
        else:
            log.info(line)

    log.info("%d country(s) checked, %d collapsed", len(countries), bad)
    return 1 if bad else 0


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

/**
 * Find Verify countries whose conversion rate has collapsed against the baseline.
 *
 * A collapse in one country on rising volume is SMS pumping in progress: the OTP
 * is delivered and billed, and nobody was ever going to enter it.
 *
 * 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';

// A country is only judged once it has this many attempts in the window.
export const MIN_ATTEMPTS = 40;

// Fractions of the service baseline, not absolute rates: any fixed threshold is
// wrong for either a 70% signup service or a 25% re-engagement one.
const COLLAPSE_RATIO = 0.35;
const WATCH_RATIO = 0.70;

/**
 * Conversion rate for one summary row, as a percentage or null. Falls back to
 * the counts so a row assembled from total_converted and total_attempts works.
 */
export function conversionRate(row) {
  const pct = row.conversion_rate_percentage;
  if (pct !== undefined && pct !== null) return Number(pct);
  const total = Number(row.total_attempts ?? 0);
  if (total <= 0) return null;
  return (100 * Number(row.total_converted ?? 0)) / total;
}

/**
 * Classify one country's summary against the service baseline. Pure, so the two
 * rules that matter -- relative to baseline, and only above a volume floor --
 * can be tested without a network. Returns [state, detail].
 */
export function verdict(row, baseline, minAttempts = MIN_ATTEMPTS) {
  const attempts = Number(row.total_attempts ?? 0);
  const country = row.country ?? '??';

  if (attempts <= 0) return ['no-traffic', 'no attempts in the window'];

  if (baseline === null || baseline === undefined || baseline <= 0) {
    return ['no-baseline',
      'the service baseline is zero or missing, so nothing can be compared ' +
      'against it: widen the window before reading this run'];
  }

  const rate = conversionRate(row);
  if (rate === null) return ['no-baseline', 'no conversion rate on the row'];

  const ratio = rate / baseline;
  const shape = `${country}: ${rate.toFixed(1)}% conversion against a ` +
                `${baseline.toFixed(1)}% baseline on ${attempts} attempts`;

  if (attempts < minAttempts) {
    return ['thin',
      `${shape}, below the ${minAttempts} attempt floor: too few to read as ` +
      'anything'];
  }

  if (ratio <= COLLAPSE_RATIO) {
    return ['collapse',
      `${shape} (${Math.round(ratio * 100)}% of baseline). The sends succeeded ` +
      'and were billed, and nobody entered the code: that is the shape of SMS ' +
      'pumping, not a broken integration.'];
  }

  if (ratio <= WATCH_RATIO) {
    return ['watch',
      `${shape} (${Math.round(ratio * 100)}% of baseline). Below the service, ` +
      'not yet at collapse: worth a second window before acting.'];
  }

  return ['healthy', `${shape} (${Math.round(ratio * 100)}% of baseline)`];
}

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

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 summary(auth, service, since, country) {
  const params = { VerifyServiceSid: service, DateCreatedAfter: since };
  if (country) params.Country = country;
  return get(auth, `${VERIFY}/Attempts/Summary`, params);
}

async function unconverted(auth, service, since, limit = 1000) {
  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);
}

/** Countries in the unconverted sweep, busiest first. */
export function countriesSeen(attempts) {
  const seen = new Map();
  for (const a of attempts) {
    if (!a.country) continue;
    seen.set(a.country, (seen.get(a.country) ?? 0) + 1);
  }
  return [...seen.entries()].sort((x, y) => y[1] - x[1]).map(([c]) => c);
}

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', '7'));
  const since = new Date(Date.now() - days * 86400000).toISOString()
    .replace(/\.\d+Z$/, 'Z');

  const baseRow = await summary(auth, service, since);
  const baseline = conversionRate(baseRow);
  console.log(`baseline ${(baseline ?? 0).toFixed(1)}% over ` +
              `${baseRow.total_attempts ?? 0} attempts`);

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

  let bad = 0;
  for (const code of countries) {
    const row = await summary(auth, service, since, code);
    if (!row.country) row.country = code;
    const [state, detail] = verdict(row, baseline);
    const line = `${state.padEnd(10)} ${detail}`;
    if (state === 'collapse' || state === 'watch') {
      if (state === 'collapse') bad += 1;
      console.warn(line);
      const hot = new Map();
      for (const a of attempts) {
        if (a.country !== code) continue;
        const p = prefixOf(a.channel_data?.to);
        hot.set(p, (hot.get(p) ?? 0) + 1);
      }
      for (const [p, n] of [...hot.entries()].sort((x, y) => y[1] - x[1]).slice(0, 3)) {
        console.warn(`  ${p} x${n} unconverted`);
      }
      console.warn(`  repair: Console > Verify > Services > ${service} > SMS: ` +
                   'enable Fraud Guard, restrict Geo Permissions to the countries ' +
                   'you serve, and add an IP-keyed Service Rate Limit');
    } else {
      console.log(line);
    }
  }

  console.log(`${countries.length} country(s) checked, ${bad} collapsed`);
  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_verify_conversion_audit import conversion_rate, prefix_of, verdict


def test_country_far_below_baseline_on_volume_is_a_collapse():
    row = {"country": "ID", "total_attempts": 812, "total_converted": 25,
           "conversion_rate_percentage": 3.1}
    state, detail = verdict(row, 64.0)
    assert state == "collapse"
    assert "pumping" in detail


def test_same_rate_on_nine_attempts_is_too_thin_to_read():
    # The volume floor is what keeps the report free of countries where four
    # people signed up this week.
    row = {"country": "MT", "total_attempts": 9, "total_converted": 0,
           "conversion_rate_percentage": 0.0}
    state, detail = verdict(row, 64.0)
    assert state == "thin"
    assert "floor" in detail


def test_judgement_is_relative_so_a_low_baseline_service_still_works():
    # 9% would trip any fixed threshold, and on this service it is normal.
    row = {"country": "BR", "total_attempts": 400, "total_converted": 36,
           "conversion_rate_percentage": 9.0}
    assert verdict(row, 11.0)[0] == "healthy"
    # Same service, a country at a fifth of that baseline.
    hit = {"country": "PK", "total_attempts": 400, "total_converted": 8,
           "conversion_rate_percentage": 2.0}
    assert verdict(hit, 11.0)[0] == "collapse"


def test_middling_country_is_watch_not_collapse():
    row = {"country": "PL", "total_attempts": 300, "total_converted": 120,
           "conversion_rate_percentage": 40.0}
    assert verdict(row, 64.0)[0] == "watch"


def test_rate_is_derived_from_the_counts_when_the_percentage_is_absent():
    assert conversion_rate({"total_attempts": 200, "total_converted": 50}) == 25.0
    assert conversion_rate({"total_attempts": 0, "total_converted": 0}) is None


def test_missing_baseline_refuses_to_judge():
    row = {"country": "US", "total_attempts": 500, "conversion_rate_percentage": 2.0}
    state, detail = verdict(row, None)
    assert state == "no-baseline"
    assert "widen the window" in detail


def test_prefix_keeps_the_leading_digits_only():
    assert prefix_of("+62 812-3456-7890") == "+628123"
    assert prefix_of(None) == "?"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { conversionRate, prefixOf, verdict } from './twilio-verify-conversion-audit.mjs';

test('country far below baseline on volume is a collapse', () => {
  const [state, detail] = verdict(
    { country: 'ID', total_attempts: 812, total_converted: 25,
      conversion_rate_percentage: 3.1 }, 64.0);
  assert.equal(state, 'collapse');
  assert.match(detail, /pumping/);
});

test('same rate on nine attempts is too thin to read', () => {
  const [state, detail] = verdict(
    { country: 'MT', total_attempts: 9, total_converted: 0,
      conversion_rate_percentage: 0 }, 64.0);
  assert.equal(state, 'thin');
  assert.match(detail, /floor/);
});

test('judgement is relative so a low baseline service still works', () => {
  assert.equal(verdict(
    { country: 'BR', total_attempts: 400, total_converted: 36,
      conversion_rate_percentage: 9.0 }, 11.0)[0], 'healthy');
  assert.equal(verdict(
    { country: 'PK', total_attempts: 400, total_converted: 8,
      conversion_rate_percentage: 2.0 }, 11.0)[0], 'collapse');
});

test('middling country is watch not collapse', () => {
  assert.equal(verdict(
    { country: 'PL', total_attempts: 300, total_converted: 120,
      conversion_rate_percentage: 40.0 }, 64.0)[0], 'watch');
});

test('rate is derived from the counts when the percentage is absent', () => {
  assert.equal(conversionRate({ total_attempts: 200, total_converted: 50 }), 25);
  assert.equal(conversionRate({ total_attempts: 0, total_converted: 0 }), null);
});

test('missing baseline refuses to judge', () => {
  const [state, detail] = verdict(
    { country: 'US', total_attempts: 500, conversion_rate_percentage: 2.0 }, null);
  assert.equal(state, 'no-baseline');
  assert.match(detail, /widen the window/);
});

test('prefix keeps the leading digits only', () => {
  assert.equal(prefixOf('+62 812-3456-7890'), '+628123');
  assert.equal(prefixOf(null), '?');
});