Verify sends SMS to a landline: 60205, or just silence

"""Find Verify traffic aimed at numbers that cannot receive an SMS.

A landline destination is not a delivery failure, it is a category error: it
returns 60205 when Lookup is on, and silently expires as an unconverted
verification when Lookup is off, which is the default.

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

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

# No SMS inbox exists behind these, whatever the carrier or the sender.
NO_SMS = {"landline", "pager", "voicemail"}

# These may or may not receive an SMS depending on the provider, which is worse
# than a clean no: the failures are intermittent and unreproducible.
UNRELIABLE = {"fixedvoip", "uan", "unknown"}


def line_type(lookup):
    """Lowercased line_type_intelligence.type, or None if the field is absent.

    The API returns camelCase values such as fixedVoip; lowercasing once here
    keeps every comparison below in one case.
    """
    lti = (lookup or {}).get("line_type_intelligence") or {}
    t = lti.get("type")
    return str(t).strip().lower() if t else None


def verdict(lookup, channel="sms"):
    """Classify one Lookup response for the channel you intend to use.

    Pure, so the rules can be tested without a network. Returns (state, detail).
    """
    if lookup is not None and lookup.get("valid") is False:
        return ("invalid",
                "Lookup says the number is not valid: it will fail on any channel")

    t = line_type(lookup)
    if t is None:
        return ("no-line-type",
                "no line_type_intelligence on the response. Either the field was "
                "not requested (Fields=line_type_intelligence) or the account is "
                "not entitled to it: do not read this as a mobile.")

    if t in NO_SMS:
        if channel == "call":
            return ("voice-ok",
                    "%s, and this verification is on the call channel: a voice "
                    "code reaches it fine" % t)
        return ("no-sms",
                "%s: there is no SMS inbox behind this number. Verify returns "
                "60205 when lookup_enabled is true, and bills a verification "
                "that expires unconverted when it is false." % t)

    if t in UNRELIABLE:
        return ("unreliable",
                "%s: SMS delivery depends entirely on the provider, so these "
                "fail intermittently and never reproduce. Offer a voice call "
                "rather than rejecting the number." % t)

    return ("mobile", "%s: can receive SMS" % t)


def guard_state(service):
    """Read lookup_enabled and skip_sms_to_landlines as the pair they are.

    Pure. The no-op combination -- skip on, lookup off -- is the one that
    convinces a team the problem is already handled.
    """
    lookup_on = bool((service or {}).get("lookup_enabled"))
    skip_on = bool((service or {}).get("skip_sms_to_landlines"))

    if skip_on and not lookup_on:
        return ("no-op",
                "skip_sms_to_landlines is true but lookup_enabled is false. The "
                "skip needs the Lookup to classify the line, so this setting "
                "does nothing at all.")
    if not lookup_on:
        return ("unguarded",
                "lookup_enabled is false: Verify cannot classify the line type, "
                "so landlines are sent to and billed in silence.")
    if not skip_on:
        return ("lookup-only",
                "lookup_enabled is true but skip_sms_to_landlines is false: you "
                "get 60205 in the logs instead of a skipped send.")
    return ("guarded", "lookup_enabled and skip_sms_to_landlines are both on")


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=1000):
    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 distinct_destinations(attempts):
    """One entry per number, with the channel it was tried on. The same number
    four times is one user trying four times, not four findings.
    """
    seen = {}
    for a in attempts:
        data = a.get("channel_data") or {}
        to = data.get("to")
        if to and to not in seen:
            seen[to] = (a.get("channel") or "sms").lower()
    return list(seen.items())


def line_type_lookup(session, e164):
    return get(session, "%s/PhoneNumbers/%s" % (LOOKUPS, e164),
               Fields="line_type_intelligence")


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 sweep")
    ap.add_argument("--max-lookups", type=int, default=60,
                    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)

    service = get(session, "%s/Services/%s" % (VERIFY, args.service))
    gstate, gdetail = guard_state(service)
    log.info("service guard: %s  %s", gstate, gdetail)
    if gstate != "guarded":
        log.warning("  repair: set LookupEnabled=true and SkipSmsToLandlines=true "
                    "on %s/Services/%s", VERIFY, args.service)

    since = (dt.datetime.now(dt.timezone.utc)
             - dt.timedelta(days=args.days)).strftime("%Y-%m-%dT%H:%M:%SZ")
    numbers = distinct_destinations(unconverted(session, args.service, since))
    if not numbers:
        log.info("no unconverted attempts in the last %d day(s)", args.days)
        return 1 if gstate != "guarded" else 0

    bad = 0
    for e164, channel in numbers[:args.max_lookups]:
        state, detail = verdict(line_type_lookup(session, e164), channel)
        line = "%-13s %s  %s" % (state, e164, detail)
        if state in ("no-sms", "invalid"):
            bad += 1
            log.warning(line)
        elif state in ("unreliable", "no-line-type"):
            log.warning(line)
        else:
            log.info(line)

    if bad:
        log.warning("  repair: gate signup on line_type_intelligence.type == "
                    "\"mobile\" and start these verifications with Channel=call "
                    "instead")

    log.info("%d number(s) sampled, %d that cannot receive SMS",
             min(len(numbers), args.max_lookups), bad)
    return 1 if (bad or gstate != "guarded") else 0


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

/**
 * Find Verify traffic aimed at numbers that cannot receive an SMS.
 *
 * A landline destination is not a delivery failure, it is a category error: it
 * returns 60205 when Lookup is on, and silently expires as an unconverted
 * verification when Lookup is off, which is the default.
 *
 * 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';

// No SMS inbox exists behind these, whatever the carrier or the sender.
const NO_SMS = new Set(['landline', 'pager', 'voicemail']);

// These may or may not receive an SMS depending on the provider, which is worse
// than a clean no: the failures are intermittent and unreproducible.
const UNRELIABLE = new Set(['fixedvoip', 'uan', 'unknown']);

/**
 * Lowercased line_type_intelligence.type, or null if the field is absent. The
 * API returns camelCase values such as fixedVoip; lowercasing once here keeps
 * every comparison below in one case.
 */
export function lineType(lookup) {
  const t = lookup?.line_type_intelligence?.type;
  return t ? String(t).trim().toLowerCase() : null;
}

/**
 * Classify one Lookup response for the channel you intend to use. Pure, so the
 * rules can be tested without a network. Returns [state, detail].
 */
export function verdict(lookup, channel = 'sms') {
  if (lookup && lookup.valid === false) {
    return ['invalid',
      'Lookup says the number is not valid: it will fail on any channel'];
  }

  const t = lineType(lookup);
  if (t === null) {
    return ['no-line-type',
      'no line_type_intelligence on the response. Either the field was not ' +
      'requested (Fields=line_type_intelligence) or the account is not ' +
      'entitled to it: do not read this as a mobile.'];
  }

  if (NO_SMS.has(t)) {
    if (channel === 'call') {
      return ['voice-ok',
        `${t}, and this verification is on the call channel: a voice code ` +
        'reaches it fine'];
    }
    return ['no-sms',
      `${t}: there is no SMS inbox behind this number. Verify returns 60205 ` +
      'when lookup_enabled is true, and bills a verification that expires ' +
      'unconverted when it is false.'];
  }

  if (UNRELIABLE.has(t)) {
    return ['unreliable',
      `${t}: SMS delivery depends entirely on the provider, so these fail ` +
      'intermittently and never reproduce. Offer a voice call rather than ' +
      'rejecting the number.'];
  }

  return ['mobile', `${t}: can receive SMS`];
}

/**
 * Read lookup_enabled and skip_sms_to_landlines as the pair they are. Pure. The
 * no-op combination -- skip on, lookup off -- is the one that convinces a team
 * the problem is already handled. Returns [state, detail].
 */
export function guardState(service) {
  const lookupOn = Boolean(service?.lookup_enabled);
  const skipOn = Boolean(service?.skip_sms_to_landlines);

  if (skipOn && !lookupOn) {
    return ['no-op',
      'skip_sms_to_landlines is true but lookup_enabled is false. The skip ' +
      'needs the Lookup to classify the line, so this setting does nothing at all.'];
  }
  if (!lookupOn) {
    return ['unguarded',
      'lookup_enabled is false: Verify cannot classify the line type, so ' +
      'landlines are sent to and billed in silence.'];
  }
  if (!skipOn) {
    return ['lookup-only',
      'lookup_enabled is true but skip_sms_to_landlines is false: you get ' +
      '60205 in the logs instead of a skipped send.'];
  }
  return ['guarded', 'lookup_enabled and skip_sms_to_landlines are both on'];
}

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

/** One entry per number: the same number four times is one user, not four findings. */
export function distinctDestinations(attempts) {
  const seen = new Map();
  for (const a of attempts) {
    const to = a.channel_data?.to;
    if (to && !seen.has(to)) seen.set(to, String(a.channel ?? 'sms').toLowerCase());
  }
  return [...seen.entries()];
}

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 serviceSid = arg('--service');
  if (!serviceSid) {
    console.error('pass --service VA...');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const days = Number(arg('--days', '7'));
  const maxLookups = Number(arg('--max-lookups', '60'));

  const service = await get(auth, `${VERIFY}/Services/${serviceSid}`);
  const [gstate, gdetail] = guardState(service);
  console.log(`service guard: ${gstate}  ${gdetail}`);
  if (gstate !== 'guarded') {
    console.warn('  repair: set LookupEnabled=true and SkipSmsToLandlines=true ' +
                 `on ${VERIFY}/Services/${serviceSid}`);
  }

  const since = new Date(Date.now() - days * 86400000).toISOString()
    .replace(/\.\d+Z$/, 'Z');
  const numbers = distinctDestinations(await unconverted(auth, serviceSid, since));
  if (numbers.length === 0) {
    console.log(`no unconverted attempts in the last ${days} day(s)`);
    process.exitCode = gstate === 'guarded' ? 0 : 1;
    return;
  }

  let bad = 0;
  for (const [e164, channel] of numbers.slice(0, maxLookups)) {
    const lookup = await get(auth, `${LOOKUPS}/PhoneNumbers/${e164}`,
                             { Fields: 'line_type_intelligence' });
    const [state, detail] = verdict(lookup, channel);
    const line = `${state.padEnd(13)} ${e164}  ${detail}`;
    if (state === 'no-sms' || state === 'invalid') { bad += 1; console.warn(line); }
    else if (state === 'unreliable' || state === 'no-line-type') console.warn(line);
    else console.log(line);
  }

  if (bad) {
    console.warn('  repair: gate signup on line_type_intelligence.type === ' +
                 '"mobile" and start these verifications with Channel=call instead');
  }

  console.log(`${Math.min(numbers.length, maxLookups)} number(s) sampled, ` +
              `${bad} that cannot receive SMS`);
  process.exitCode = (bad || gstate !== 'guarded') ? 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_landline_audit import guard_state, line_type, verdict


def test_landline_on_the_sms_channel_is_the_finding():
    state, detail = verdict({"line_type_intelligence": {"type": "landline"}})
    assert state == "no-sms"
    assert "60205" in detail


def test_the_same_landline_on_a_voice_verification_is_fine():
    state, _ = verdict({"line_type_intelligence": {"type": "landline"}},
                       channel="call")
    assert state == "voice-ok"


def test_fixed_voip_is_neither_a_pass_nor_a_rejection():
    # Camel case from the API, matched case-insensitively.
    state, detail = verdict({"line_type_intelligence": {"type": "fixedVoip"}})
    assert state == "unreliable"
    assert "voice call" in detail


def test_a_response_with_no_line_type_is_not_a_mobile():
    state, detail = verdict({"valid": True})
    assert state == "no-line-type"
    assert "Fields=line_type_intelligence" in detail


def test_mobile_passes():
    assert verdict({"line_type_intelligence": {"type": "mobile"}})[0] == "mobile"
    assert line_type({"line_type_intelligence": {"type": "  Mobile "}}) == "mobile"


def test_skip_without_lookup_is_a_setting_that_does_nothing():
    state, detail = guard_state({"lookup_enabled": False,
                                 "skip_sms_to_landlines": True})
    assert state == "no-op"
    assert "does nothing" in detail


def test_both_settings_on_is_the_only_guarded_state():
    assert guard_state({"lookup_enabled": True,
                        "skip_sms_to_landlines": True})[0] == "guarded"
    assert guard_state({"lookup_enabled": True,
                        "skip_sms_to_landlines": False})[0] == "lookup-only"
    assert guard_state({})[0] == "unguarded"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { guardState, lineType, verdict } from './twilio-verify-landline-audit.mjs';

test('landline on the sms channel is the finding', () => {
  const [state, detail] = verdict({ line_type_intelligence: { type: 'landline' } });
  assert.equal(state, 'no-sms');
  assert.match(detail, /60205/);
});

test('the same landline on a voice verification is fine', () => {
  assert.equal(
    verdict({ line_type_intelligence: { type: 'landline' } }, 'call')[0],
    'voice-ok');
});

test('fixed voip is neither a pass nor a rejection', () => {
  const [state, detail] = verdict({ line_type_intelligence: { type: 'fixedVoip' } });
  assert.equal(state, 'unreliable');
  assert.match(detail, /voice call/);
});

test('a response with no line type is not a mobile', () => {
  const [state, detail] = verdict({ valid: true });
  assert.equal(state, 'no-line-type');
  assert.match(detail, /Fields=line_type_intelligence/);
});

test('mobile passes', () => {
  assert.equal(verdict({ line_type_intelligence: { type: 'mobile' } })[0], 'mobile');
  assert.equal(lineType({ line_type_intelligence: { type: '  Mobile ' } }), 'mobile');
});

test('skip without lookup is a setting that does nothing', () => {
  const [state, detail] = guardState({ lookup_enabled: false,
                                       skip_sms_to_landlines: true });
  assert.equal(state, 'no-op');
  assert.match(detail, /does nothing/);
});

test('both settings on is the only guarded state', () => {
  assert.equal(guardState({ lookup_enabled: true,
                            skip_sms_to_landlines: true })[0], 'guarded');
  assert.equal(guardState({ lookup_enabled: true,
                            skip_sms_to_landlines: false })[0], 'lookup-only');
  assert.equal(guardState({})[0], 'unguarded');
});