60202: a verification that burned all five check attempts

"""Report Verify verifications that spent all five checks and died.

A verification allows five checks. A handler that fires on every keystroke, or
a form that submits twice, spends them before the user finishes typing, and the
verification moves to max_attempts_reached for the rest of its lifetime.

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 logging
import os
import sys
from datetime import datetime, timedelta, timezone

import requests

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

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

# Fixed by the platform: five checks per verification, and the verification lives
# ten minutes. Both numbers are the reason a burned one cannot be recovered.
MAX_CHECKS = 5
TTL_SECONDS = 600


def parse_time(value):
    """Parse a Verify timestamp into an aware datetime, or None.

    fromisoformat did not accept a trailing Z until 3.11 and every timestamp
    Verify returns has one, so the swap has to happen here rather than in the
    caller.
    """
    s = str(value or "").strip()
    if not s:
        return None
    if s.endswith("Z"):
        s = s[:-1] + "+00:00"
    try:
        dt = datetime.fromisoformat(s)
    except ValueError:
        return None
    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)


def age_seconds(value, now):
    dt = parse_time(value)
    return None if dt is None else (now - dt).total_seconds()


def verdict(http_status, verification, now, ttl_seconds=TTL_SECONDS):
    """Classify one verification lookup.

    `http_status` matters as much as the body: Verify soft-deletes a verification
    once it is approved, canceled or expired, so a 404 means it resolved rather
    than that anything went wrong. Pure, so that rule can be tested without a
    network. Returns (state, detail).
    """
    if http_status == 404:
        return ("resolved",
                "404: the verification is soft deleted, which Verify does once "
                "it is approved, canceled or expired. Nothing is stuck.")

    body = verification or {}
    status = str(body.get("status") or "").strip().lower()

    if status == "max_attempts_reached":
        age = age_seconds(body.get("date_created"), now)
        if age is None:
            return ("burned",
                    "all %d checks spent. date_created is unreadable, so "
                    "whether the lifetime has run out cannot be told from here."
                    % MAX_CHECKS)
        remaining = ttl_seconds - age
        if remaining > 0:
            return ("burned-live",
                    "all %d checks spent %ds ago. Every further check returns "
                    "60202 for another %ds, and someone is looking at that "
                    "screen now." % (MAX_CHECKS, int(age), int(remaining)))
        return ("burned-cold",
                "all %d checks spent %ds ago, past the %ds lifetime. Nobody is "
                "stuck on it; it counts towards the rate."
                % (MAX_CHECKS, int(age), ttl_seconds))

    if status == "pending":
        return ("pending", "open, checks still available")
    if status in ("approved", "canceled"):
        return (status, "closed as " + status)
    return ("unknown", "status %r is not one this script recognises"
            % (body.get("status"),))


def fetch(session, url, **params):
    """GET returning (status_code, body). 404 is data here, not an error."""
    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)
    if r.status_code == 404:
        return (404, {})
    r.raise_for_status()
    return (r.status_code, r.json())


def get(session, url, **params):
    status, body = fetch(session, url, **params)
    if status == 404:
        raise SystemExit("404 from %s: check the SID" % url)
    return body


def page(session, url, field, **params):
    """Walk a Verify v2 list. Paging lives in meta.next_page_url."""
    out = []
    while url:
        body = get(session, url, **params)
        out.extend(body.get(field, []))
        url, params = (body.get("meta") or {}).get("next_page_url"), {}
    return out


def verification_sids(session, service_sid, since, limit):
    """Distinct verification SIDs seen in the attempts list for a window.

    There is no list endpoint for verifications, so the attempts are the only
    read-only way to find out which ones existed. Order is preserved so the
    limit takes the oldest in the window rather than an arbitrary slice.
    """
    seen, out = set(), []
    for attempt in page(session, VERIFY + "/Attempts", "attempts",
                        VerifyServiceSid=service_sid, DateCreatedAfter=since,
                        PageSize=100):
        sid = attempt.get("verification_sid")
        if sid and sid not in seen:
            seen.add(sid)
            out.append(sid)
            if len(out) >= limit:
                break
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--service", action="append", default=[],
                    help="Verify Service SID; repeatable. Default: every service")
    ap.add_argument("--hours", type=int, default=24,
                    help="how far back to look for verifications")
    ap.add_argument("--max-verifications", type=int, default=500,
                    help="stop after this many per service")
    ap.add_argument("--burn-rate", type=float, default=2.0,
                    help="percent burned above which this is a client bug")
    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)

    now = datetime.now(timezone.utc)
    since = (now - timedelta(hours=args.hours)).strftime("%Y-%m-%dT%H:%M:%SZ")

    if args.service:
        services = [{"sid": s, "friendly_name": s} for s in args.service]
    else:
        services = page(session, VERIFY + "/Services", "services", PageSize=50)
    if not services:
        log.info("no Verify services on this account")
        return 0

    inspected = burned = live = 0
    for svc in services:
        sid = svc.get("sid")
        for ve in verification_sids(session, sid, since, args.max_verifications):
            status, body = fetch(
                session, "%s/Services/%s/Verifications/%s" % (VERIFY, sid, ve))
            state, detail = verdict(status, body, now)
            inspected += 1
            if not state.startswith("burned"):
                continue
            burned += 1
            log.warning("%-12s %s  %s", state, ve, detail)
            if state == "burned-live":
                live += 1
                log.warning("  repair now: POST %s/Services/%s/Verifications/%s "
                            "with Status=canceled, then start a fresh "
                            "verification for that user", VERIFY, sid, ve)

    if not inspected:
        log.info("no verifications in the last %d hour(s)", args.hours)
        return 0

    rate = 100.0 * burned / inspected
    log.info("%d verification(s) inspected, %d burned (%.1f%%), %d still inside "
             "their lifetime", inspected, burned, rate, live)
    if rate > args.burn_rate:
        log.warning("above %.1f%%: debounce the check call and submit only on a "
                    "complete code. 60202 is terminal, not retryable.",
                    args.burn_rate)
        return 1
    return 1 if live else 0


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

/**
 * Report Verify verifications that spent all five checks and died.
 *
 * A verification allows five checks. A handler that fires on every keystroke,
 * or a form that submits twice, spends them before the user finishes typing,
 * and the verification moves to max_attempts_reached for the rest of its
 * lifetime.
 *
 * 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';

// Fixed by the platform: five checks per verification, ten minute lifetime.
export const MAX_CHECKS = 5;
export const TTL_SECONDS = 600;

/** Parse a Verify timestamp into epoch milliseconds, or null. */
export function parseTime(value) {
  const s = String(value ?? '').trim();
  if (!s) return null;
  const ms = Date.parse(s);
  return Number.isFinite(ms) ? ms : null;
}

export function ageSeconds(value, nowMs) {
  const ms = parseTime(value);
  return ms === null ? null : (nowMs - ms) / 1000;
}

/**
 * Classify one verification lookup. `httpStatus` matters as much as the body:
 * Verify soft-deletes a verification once it is approved, canceled or expired,
 * so a 404 means it resolved rather than that anything went wrong. Pure, so
 * that rule can be tested without a network. Returns [state, detail].
 */
export function verdict(httpStatus, verification, nowMs, ttlSeconds = TTL_SECONDS) {
  if (httpStatus === 404) {
    return ['resolved',
      '404: the verification is soft deleted, which Verify does once it is ' +
      'approved, canceled or expired. Nothing is stuck.'];
  }

  const body = verification ?? {};
  const status = String(body.status ?? '').trim().toLowerCase();

  if (status === 'max_attempts_reached') {
    const age = ageSeconds(body.date_created, nowMs);
    if (age === null) {
      return ['burned',
        `all ${MAX_CHECKS} checks spent. date_created is unreadable, so ` +
        'whether the lifetime has run out cannot be told from here.'];
    }
    const remaining = ttlSeconds - age;
    if (remaining > 0) {
      return ['burned-live',
        `all ${MAX_CHECKS} checks spent ${Math.trunc(age)}s ago. Every further ` +
        `check returns 60202 for another ${Math.trunc(remaining)}s, and ` +
        'someone is looking at that screen now.'];
    }
    return ['burned-cold',
      `all ${MAX_CHECKS} checks spent ${Math.trunc(age)}s ago, past the ` +
      `${ttlSeconds}s lifetime. Nobody is stuck on it; it counts towards the rate.`];
  }

  if (status === 'pending') return ['pending', 'open, checks still available'];
  if (status === 'approved' || status === 'canceled') {
    return [status, `closed as ${status}`];
  }
  return ['unknown',
    `status ${JSON.stringify(body.status)} is not one this script recognises`];
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

/** GET returning [status, body]. 404 is data here, not an error. */
async function fetchJson(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.status === 404) return [404, {}];
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return [res.status, await res.json()];
}

async function get(auth, url, params = {}) {
  const [status, body] = await fetchJson(auth, url, params);
  if (status === 404) throw new Error(`404 from ${url}: check the SID`);
  return body;
}

/** Walk a Verify v2 list. Paging lives in meta.next_page_url. */
async function page(auth, url, field, params = {}) {
  const out = [];
  let next = url;
  let p = params;
  while (next) {
    const body = await get(auth, next, p);
    out.push(...(body[field] ?? []));
    next = body.meta?.next_page_url ?? null;
    p = {};
  }
  return out;
}

/** Distinct verification SIDs seen in the attempts list for a window. */
export async function verificationSids(auth, serviceSid, since, limit) {
  const seen = new Set();
  const attempts = await page(auth, `${VERIFY}/Attempts`, 'attempts', {
    VerifyServiceSid: serviceSid, DateCreatedAfter: since, PageSize: 100,
  });
  for (const attempt of attempts) {
    if (attempt.verification_sid) seen.add(attempt.verification_sid);
    if (seen.size >= limit) break;
  }
  return [...seen];
}

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 hoursArg = process.argv.indexOf('--hours');
  const hours = hoursArg === -1 ? 24 : Number(process.argv[hoursArg + 1]) || 24;
  const nowMs = Date.now();
  const since = `${new Date(nowMs - hours * 3600 * 1000).toISOString().slice(0, 19)}Z`;

  const services = await page(auth, `${VERIFY}/Services`, 'services', { PageSize: 50 });
  if (services.length === 0) {
    console.log('no Verify services on this account');
    return;
  }

  let inspected = 0;
  let burned = 0;
  let live = 0;
  for (const svc of services) {
    for (const ve of await verificationSids(auth, svc.sid, since, 500)) {
      const [status, body] = await fetchJson(
        auth, `${VERIFY}/Services/${svc.sid}/Verifications/${ve}`);
      const [state, detail] = verdict(status, body, nowMs);
      inspected += 1;
      if (!state.startsWith('burned')) continue;
      burned += 1;
      console.warn(`${state.padEnd(12)} ${ve}  ${detail}`);
      if (state === 'burned-live') {
        live += 1;
        console.warn(`  repair now: POST ${VERIFY}/Services/${svc.sid}/` +
                     `Verifications/${ve} with Status=canceled, then start a ` +
                     'fresh verification for that user');
      }
    }
  }

  if (inspected === 0) {
    console.log(`no verifications in the last ${hours} hour(s)`);
    return;
  }

  const rate = (100 * burned) / inspected;
  console.log(`${inspected} verification(s) inspected, ${burned} burned ` +
              `(${rate.toFixed(1)}%), ${live} still inside their lifetime`);
  if (rate > 2) {
    console.warn('above 2.0%: debounce the check call and submit only on a ' +
                 'complete code. 60202 is terminal, not retryable.');
  }
  process.exitCode = rate > 2 || live ? 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 datetime import datetime, timedelta, timezone

from twilio_verify_check_attempts_audit import age_seconds, parse_time, verdict

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


def iso(seconds_ago):
    return (NOW - timedelta(seconds=seconds_ago)).strftime("%Y-%m-%dT%H:%M:%SZ")


def test_404_is_resolved_not_an_error():
    state, detail = verdict(404, None, NOW)
    assert state == "resolved"
    assert "soft deleted" in detail


def test_burned_inside_the_lifetime_is_someone_waiting_now():
    state, detail = verdict(200, {"status": "max_attempts_reached",
                                  "date_created": iso(120)}, NOW)
    assert state == "burned-live"
    assert "another 480s" in detail


def test_burned_after_the_lifetime_is_only_a_statistic():
    state, detail = verdict(200, {"status": "max_attempts_reached",
                                  "date_created": iso(3600)}, NOW)
    assert state == "burned-cold"
    assert "Nobody is stuck" in detail


def test_burned_with_an_unreadable_clock_is_still_burned():
    state, detail = verdict(200, {"status": "max_attempts_reached",
                                  "date_created": "not a date"}, NOW)
    assert state == "burned"
    assert "unreadable" in detail


def test_pending_and_approved_are_left_alone():
    assert verdict(200, {"status": "pending"}, NOW)[0] == "pending"
    assert verdict(200, {"status": "approved"}, NOW)[0] == "approved"


def test_an_unrecognised_status_is_reported_rather_than_assumed_healthy():
    state, detail = verdict(200, {"status": "expired"}, NOW)
    assert state == "unknown"
    assert "expired" in detail


def test_timestamps_with_a_trailing_z_parse_on_3_9():
    assert parse_time("2026-03-04T11:58:00Z") is not None
    assert parse_time("") is None
    assert age_seconds(iso(60), NOW) == 60

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { ageSeconds, parseTime, verdict } from './twilio-verify-check-attempts-audit.mjs';

const NOW = Date.parse('2026-03-04T12:00:00Z');
const iso = (secondsAgo) =>
  `${new Date(NOW - secondsAgo * 1000).toISOString().slice(0, 19)}Z`;

test('404 is resolved not an error', () => {
  const [state, detail] = verdict(404, null, NOW);
  assert.equal(state, 'resolved');
  assert.match(detail, /soft deleted/);
});

test('burned inside the lifetime is someone waiting now', () => {
  const [state, detail] = verdict(
    200, { status: 'max_attempts_reached', date_created: iso(120) }, NOW);
  assert.equal(state, 'burned-live');
  assert.match(detail, /another 480s/);
});

test('burned after the lifetime is only a statistic', () => {
  const [state, detail] = verdict(
    200, { status: 'max_attempts_reached', date_created: iso(3600) }, NOW);
  assert.equal(state, 'burned-cold');
  assert.match(detail, /Nobody is stuck/);
});

test('burned with an unreadable clock is still burned', () => {
  const [state, detail] = verdict(
    200, { status: 'max_attempts_reached', date_created: 'not a date' }, NOW);
  assert.equal(state, 'burned');
  assert.match(detail, /unreadable/);
});

test('pending and approved are left alone', () => {
  assert.equal(verdict(200, { status: 'pending' }, NOW)[0], 'pending');
  assert.equal(verdict(200, { status: 'approved' }, NOW)[0], 'approved');
});

test('an unrecognised status is reported rather than assumed healthy', () => {
  const [state, detail] = verdict(200, { status: 'expired' }, NOW);
  assert.equal(state, 'unknown');
  assert.match(detail, /expired/);
});

test('timestamps parse and age is measured in seconds', () => {
  assert.notEqual(parseTime('2026-03-04T11:58:00Z'), null);
  assert.equal(parseTime(''), null);
  assert.equal(ageSeconds(iso(60), NOW), 60);
});