a ten hour validity period delivers passcodes nobody wants

"""Report Messaging Services whose validity period is far longer than the
traffic they carry can use.

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

import requests

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

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

# The value a Messaging Service is created with: ten hours.
DEFAULT_VALIDITY = 36000


def queue_seconds(message):
    """How long a message waited between being accepted and being sent.

    Both timestamps are RFC 2822, so they are parsed rather than sliced: a fixed
    length slice of that format reads the same for every message ever sent, and
    a backlog measured that way looks like no backlog at all.

    Pure. Returns None when either timestamp is missing or unparseable, because
    "not measured" and "waited nothing" are different facts.
    """
    def parse(value):
        if not value:
            return None
        try:
            return email.utils.parsedate_to_datetime(str(value))
        except (TypeError, ValueError):
            return None

    created, sent = parse(message.get("date_created")), parse(message.get("date_sent"))
    if created is None or sent is None:
        return None
    return (sent - created).total_seconds()


def verdict(service, latency=None, time_critical=None):
    """Weigh a service's validity period against what its queue actually did.

    `latency` is {"sampled": n, "late": k, "worst": seconds} or None when the
    Messages window produced no rows for this service. `time_critical` is True,
    False or None: the API has no field for what a service carries, so this is
    declared rather than guessed.

    Pure, so the ranking can be tested without a network.
    Returns (state, detail).
    """
    raw = service.get("validity_period")
    if raw is None:
        return ("unknown", "validity_period was not read, so nothing can be said "
                           "about the deadline this service enforces.")
    period = int(raw)
    late = (latency or {}).get("late") or 0
    worst = int((latency or {}).get("worst") or 0)

    if period < DEFAULT_VALIDITY:
        return ("capped",
                "validity_period is %ds rather than the %ds default. The failure "
                "at this end is 30036, messages expiring in the queue, so keep "
                "it above the wait you actually measure."
                % (period, DEFAULT_VALIDITY))

    if time_critical is False:
        return ("bulk",
                "the ten hour default, on traffic declared not time critical. "
                "That is what the default is for.")

    if late:
        return ("too-long",
                "%d of %d sampled message(s) waited past the threshold, worst "
                "%ds, under a %ds ceiling. A passcode behind that queue is "
                "delivered rather than dropped, hours after it was any use."
                % (late, (latency or {}).get("sampled") or 0, worst, period))

    if time_critical:
        return ("latent",
                "declared time critical and still carrying the ten hour default. "
                "Nothing is arriving late in this window, and nothing stops it "
                "during the next backlog either.")

    return ("undeclared",
            "the ten hour default, and this script cannot tell what the service "
            "carries. Declare it with --time-critical or --bulk.")


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_services(session, limit):
    url = "%s/Services" % MESSAGING
    params = {"PageSize": 100}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("services", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def sample_latency(session, account, days, threshold, max_messages):
    """Queue wait per Messaging Service, from the Messages list.

    The list has no status filter and no error code filter, so the window is
    paged and everything is computed here. Rows with no messaging_service_sid
    were sent with a bare From and are governed by no service setting.
    """
    since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    url = "%s/Accounts/%s/Messages.json" % (BASE, account)
    params = {"DateSent>": since, "PageSize": 1000}
    stats = collections.defaultdict(lambda: {"sampled": 0, "late": 0, "worst": 0})
    seen = 0
    while url and seen < max_messages:
        page = get(session, url, **params)
        rows = page.get("messages", [])
        seen += len(rows)
        for m in rows:
            sid = m.get("messaging_service_sid")
            waited = queue_seconds(m)
            if not sid or waited is None:
                continue
            s = stats[sid]
            s["sampled"] += 1
            s["worst"] = max(s["worst"], waited)
            if waited > threshold:
                s["late"] += 1
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return stats


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--time-critical", action="append", default=[],
                    help="Messaging Service SID carrying passcodes or alerts")
    ap.add_argument("--bulk", action="append", default=[],
                    help="Messaging Service SID carrying campaign traffic")
    ap.add_argument("--days", type=int, default=7, help="latency window in days")
    ap.add_argument("--late-after", type=int, default=120,
                    help="seconds of queue wait that counts as late")
    ap.add_argument("--max-messages", type=int, default=20000,
                    help="stop paging the Messages list after this many rows")
    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)

    declared = {sid: True for sid in args.time_critical}
    declared.update({sid: False for sid in args.bulk})

    services = list_services(session, 200)
    stats = sample_latency(session, account, args.days, args.late_after,
                           args.max_messages)

    bad = 0
    for svc in services:
        sid = svc.get("sid")
        state, detail = verdict(svc, stats.get(sid), declared.get(sid))
        label = svc.get("friendly_name") or sid
        line = "%-12s %s  %s" % (state, label, detail)
        if state in ("capped", "bulk"):
            log.info(line)
            continue
        if state in ("unknown", "undeclared"):
            log.warning(line)
            continue
        bad += 1
        log.warning(line)
        log.warning("  repair: POST %s/Services/%s ValidityPeriod=300 for time "
                    "critical traffic, and add senders if the measured wait is "
                    "already longer than the new ceiling, or you have chosen to "
                    "fail fast rather than late.", MESSAGING, sid)

    log.info("%d service(s), %d with a ten hour ceiling over time critical traffic",
             len(services), bad)
    return 1 if bad else 0


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

/**
 * Report Messaging Services whose validity period is far longer than the traffic
 * they carry can use.
 *
 * 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`;
const MESSAGING = 'https://messaging.twilio.com/v1';

// The value a Messaging Service is created with: ten hours.
const DEFAULT_VALIDITY = 36000;

/**
 * How long a message waited between being accepted and being sent. Both
 * timestamps are RFC 2822, which Date.parse handles; null when either is
 * missing or unparseable, because "not measured" and "waited nothing" are
 * different facts. Pure.
 */
export function queueSeconds(message) {
  const created = Date.parse(message.date_created ?? '');
  const sent = Date.parse(message.date_sent ?? '');
  if (Number.isNaN(created) || Number.isNaN(sent)) return null;
  return (sent - created) / 1000;
}

/**
 * Weigh a service's validity period against what its queue actually did.
 * `latency` is {sampled, late, worst} or null; `timeCritical` is true, false or
 * null, because the API has no field for what a service carries. Pure.
 * Returns [state, detail].
 */
export function verdict(service, latency = null, timeCritical = null) {
  const raw = service.validity_period;
  if (raw === null || raw === undefined) {
    return ['unknown',
      'validity_period was not read, so nothing can be said about the deadline ' +
      'this service enforces.'];
  }
  const period = Number(raw);
  const late = latency?.late ?? 0;
  const worst = Math.round(latency?.worst ?? 0);

  if (period < DEFAULT_VALIDITY) {
    return ['capped',
      `validity_period is ${period}s rather than the ${DEFAULT_VALIDITY}s ` +
      'default. The failure at this end is 30036, messages expiring in the ' +
      'queue, so keep it above the wait you actually measure.'];
  }

  if (timeCritical === false) {
    return ['bulk',
      'the ten hour default, on traffic declared not time critical. That is ' +
      'what the default is for.'];
  }

  if (late) {
    return ['too-long',
      `${late} of ${latency?.sampled ?? 0} sampled message(s) waited past the ` +
      `threshold, worst ${worst}s, under a ${period}s ceiling. A passcode ` +
      'behind that queue is delivered rather than dropped, hours after it was ' +
      'any use.'];
  }

  if (timeCritical) {
    return ['latent',
      'declared time critical and still carrying the ten hour default. Nothing ' +
      'is arriving late in this window, and nothing stops it during the next ' +
      'backlog either.'];
  }

  return ['undeclared',
    'the ten hour default, and this script cannot tell what the service ' +
    'carries. Declare it with --time-critical or --bulk.'];
}

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

export async function listServices(auth, limit = 200) {
  let url = `${MESSAGING}/Services`;
  let params = { PageSize: 100 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.services ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

/** Queue wait per Messaging Service. Rows with no service sid used a bare From. */
async function sampleLatency(auth, account, days, threshold, maxMessages) {
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);
  let url = `${BASE}/Accounts/${account}/Messages.json`;
  let params = { 'DateSent>': since, PageSize: 1000 };
  const stats = new Map();
  let seen = 0;
  while (url && seen < maxMessages) {
    const page = await get(auth, url, params);
    const rows = page.messages ?? [];
    seen += rows.length;
    for (const m of rows) {
      const sid = m.messaging_service_sid;
      const waited = queueSeconds(m);
      if (!sid || waited === null) continue;
      const s = stats.get(sid) ?? { sampled: 0, late: 0, worst: 0 };
      s.sampled += 1;
      s.worst = Math.max(s.worst, waited);
      if (waited > threshold) s.late += 1;
      stats.set(sid, s);
    }
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return stats;
}

function repeatedFlag(name) {
  const out = [];
  process.argv.forEach((a, i) => { if (a === name) out.push(process.argv[i + 1]); });
  return out.filter(Boolean);
}

function flagValue(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 declared = new Map();
  for (const sid of repeatedFlag('--time-critical')) declared.set(sid, true);
  for (const sid of repeatedFlag('--bulk')) declared.set(sid, false);

  const services = await listServices(auth);
  const stats = await sampleLatency(auth, account, flagValue('--days', 7),
                                    flagValue('--late-after', 120),
                                    flagValue('--max-messages', 20000));

  let bad = 0;
  for (const svc of services) {
    const [state, detail] = verdict(svc, stats.get(svc.sid) ?? null,
                                    declared.has(svc.sid) ? declared.get(svc.sid) : null);
    const line = `${state.padEnd(12)} ${svc.friendly_name ?? svc.sid}  ${detail}`;
    if (state === 'capped' || state === 'bulk') { console.log(line); continue; }
    if (state === 'unknown' || state === 'undeclared') { console.warn(line); continue; }
    bad += 1;
    console.warn(line);
    console.warn(`  repair: POST ${MESSAGING}/Services/${svc.sid} ValidityPeriod=300 ` +
                 'for time critical traffic, and add senders if the measured wait ' +
                 'is already longer than the new ceiling, or you have chosen to ' +
                 'fail fast rather than late.');
  }

  console.log(`${services.length} service(s), ${bad} with a ten hour ceiling ` +
              'over time critical traffic');
  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_validity_ceiling_audit import queue_seconds, verdict

DEFAULT = {"sid": "MG1", "friendly_name": "notifications", "validity_period": 36000}
TIGHT = {"sid": "MG2", "friendly_name": "passcodes", "validity_period": 300}


def test_queue_wait_is_parsed_from_rfc_2822_not_sliced():
    waited = queue_seconds({"date_created": "Mon, 24 Aug 2026 09:00:00 +0000",
                            "date_sent": "Mon, 24 Aug 2026 09:04:00 +0000"})
    assert waited == 240


def test_a_message_that_was_never_sent_measures_nothing():
    # None rather than 0: not measured and waited nothing are different facts.
    assert queue_seconds({"date_created": "Mon, 24 Aug 2026 09:00:00 +0000"}) is None
    assert queue_seconds({"date_created": "nonsense", "date_sent": "also nonsense"}) is None


def test_measured_late_deliveries_under_the_default_are_the_finding():
    state, detail = verdict(DEFAULT, {"sampled": 400, "late": 37, "worst": 5400}, True)
    assert state == "too-long"
    assert "5400s" in detail


def test_late_deliveries_outrank_a_missing_declaration():
    assert verdict(DEFAULT, {"sampled": 10, "late": 1, "worst": 900})[0] == "too-long"


def test_the_default_is_correct_for_traffic_declared_bulk():
    state, _ = verdict(DEFAULT, {"sampled": 900, "late": 90, "worst": 7200}, False)
    assert state == "bulk"


def test_time_critical_traffic_at_the_default_is_latent_even_with_a_clean_window():
    state, detail = verdict(DEFAULT, {"sampled": 500, "late": 0, "worst": 12}, True)
    assert state == "latent"
    assert "next backlog" in detail


def test_an_undeclared_service_asks_rather_than_guesses():
    assert verdict(DEFAULT, {"sampled": 500, "late": 0, "worst": 9})[0] == "undeclared"


def test_a_shorter_ceiling_points_at_the_other_failure():
    state, detail = verdict(TIGHT, {"sampled": 500, "late": 0, "worst": 9}, True)
    assert state == "capped"
    assert "30036" in detail


def test_an_unread_validity_period_is_never_guessed_at():
    assert verdict({"sid": "MG3"}, None, True)[0] == "unknown"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { queueSeconds, verdict } from './twilio-validity-ceiling-audit.mjs';

const DEFAULT = { sid: 'MG1', friendly_name: 'notifications', validity_period: 36000 };
const TIGHT = { sid: 'MG2', friendly_name: 'passcodes', validity_period: 300 };

test('queue wait is parsed from rfc 2822 not sliced', () => {
  assert.equal(queueSeconds({ date_created: 'Mon, 24 Aug 2026 09:00:00 +0000',
                              date_sent: 'Mon, 24 Aug 2026 09:04:00 +0000' }), 240);
});

test('a message that was never sent measures nothing', () => {
  assert.equal(queueSeconds({ date_created: 'Mon, 24 Aug 2026 09:00:00 +0000' }), null);
  assert.equal(queueSeconds({ date_created: 'nonsense', date_sent: 'also nonsense' }), null);
});

test('measured late deliveries under the default are the finding', () => {
  const [state, detail] = verdict(DEFAULT, { sampled: 400, late: 37, worst: 5400 }, true);
  assert.equal(state, 'too-long');
  assert.match(detail, /5400s/);
});

test('late deliveries outrank a missing declaration', () => {
  assert.equal(verdict(DEFAULT, { sampled: 10, late: 1, worst: 900 })[0], 'too-long');
});

test('the default is correct for traffic declared bulk', () => {
  assert.equal(verdict(DEFAULT, { sampled: 900, late: 90, worst: 7200 }, false)[0], 'bulk');
});

test('time critical traffic at the default is latent even with a clean window', () => {
  const [state, detail] = verdict(DEFAULT, { sampled: 500, late: 0, worst: 12 }, true);
  assert.equal(state, 'latent');
  assert.match(detail, /next backlog/);
});

test('an undeclared service asks rather than guesses', () => {
  assert.equal(verdict(DEFAULT, { sampled: 500, late: 0, worst: 9 })[0], 'undeclared');
});

test('a shorter ceiling points at the other failure', () => {
  const [state, detail] = verdict(TIGHT, { sampled: 500, late: 0, worst: 9 }, true);
  assert.equal(state, 'capped');
  assert.match(detail, /30036/);
});

test('an unread validity period is never guessed at', () => {
  assert.equal(verdict({ sid: 'MG3' }, null, true)[0], 'unknown');
});