a TwiML response over 64 kB drops the call with 11750

"""Report Twilio webhooks whose response exceeds the 64 kB TwiML limit (11750).

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 datetime as dt
import logging
import os
import sys
from urllib.parse import urlsplit

import requests

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

MONITOR = "https://monitor.twilio.com/v1"

BODY_TOO_LARGE = 11750

# The cap Twilio applies to a TwiML response, in bytes.
LIMIT = 64 * 1024

# Alerts are retained 30 days. A longer window is not more history, it is the
# same history under a label that makes the report look more thorough.
MAX_DAYS = 30

# Markers of a framework error page: the far commoner cause of 11750 than a
# genuinely large document.
TRACE_MARKERS = (
    "traceback (most recent call last)",
    "stack trace",
    "stacktrace",
    "whoops! there was an error",
    "werkzeug debugger",
    "actiondispatch",
)


def code_of(alert):
    """Read error_code off an alert as an integer, or None.

    The Monitor API hands this back as a string while the Messages list hands
    back a number for the same concept, and a check written for one and pointed
    at the other matches nothing and reports a healthy account.
    """
    raw = alert.get("error_code")
    if raw is None or raw == "":
        return None
    try:
        return int(raw)
    except (TypeError, ValueError):
        return None


def endpoint_of(url):
    """Lowercase host plus path, for grouping. Query string dropped."""
    if not url:
        return ""
    parts = urlsplit(str(url).strip())
    host = (parts.hostname or "").lower()
    if not host:
        return str(url).strip().lower().rstrip("/")
    return host + (parts.path or "").rstrip("/")


def byte_length(text):
    """Size in bytes, which is the unit the limit is expressed in.

    len() on a string counts characters. A document with accented or non-Latin
    text in <Say> is larger on the wire than its length, so a character count
    reads a failing response as comfortably inside the cap.
    """
    return len(str(text or "").encode("utf-8"))


def classify_body(body):
    """Say what the oversized response actually was. Pure.

    The verdict keys on what the body is rather than on how long it is, because
    response_body is stored with a size limit of its own: measuring it gives a
    floor, never the size of the response Twilio refused.

    Returns (state, detail).
    """
    raw = "" if body is None else str(body)
    size = byte_length(raw)

    if not raw.strip():
        return ("no-body",
                "the single-alert fetch returned nothing, so the cause cannot "
                "be read from here. Reproduce the request against the handler "
                "and measure what it writes.")

    low = raw.lstrip().lower()
    if (low.startswith("<!doctype html") or low.startswith("<html")
            or "<html" in low[:2000]):
        return ("error-page",
                "an HTML page, not TwiML: at least %d bytes of framework debug "
                "output. The size is a symptom; the handler threw." % size)

    if any(m in low for m in TRACE_MARKERS):
        return ("stack-trace",
                "a stack trace, at least %d bytes of it. Debug output is still "
                "on in production and every unhandled exception returns an "
                "essay." % size)

    if "<response" in low:
        if size >= LIMIT:
            return ("oversized-twiml",
                    "real TwiML, %d bytes, over the %d byte cap. This one needs "
                    "splitting rather than fixing." % (size, LIMIT))
        return ("twiml-truncated",
                "real TwiML. The stored copy is %d bytes, under the cap, but "
                "response_body is truncated: that is a floor, not the size of "
                "the response." % size)

    return ("not-twiml",
            "at least %d bytes of something that is neither TwiML nor a "
            "recognisable error page. Read the first line of it." % size)


def group(alerts, code=BODY_TOO_LARGE):
    """Bucket alerts with one error code by endpoint. Pure."""
    out = {}
    for a in alerts:
        if code_of(a) != code:
            continue
        key = endpoint_of(a.get("request_url"))
        row = out.setdefault(key, {"alerts": 0, "sids": [], "first": None,
                                   "last": None})
        row["alerts"] += 1
        if len(row["sids"]) < 3:
            row["sids"].append(a.get("sid"))
        when = a.get("date_generated") or ""
        if when:
            row["first"] = when if row["first"] is None else min(row["first"], when)
            row["last"] = when if row["last"] is None else max(row["last"], when)
    return out


REPAIRS = {
    "error-page": "fix the exception, and turn debug pages off in production so "
                  "a failure returns a short 500 rather than a rendered page",
    "stack-trace": "disable debug output in production and return a small TwiML "
                   "document from the error branch",
    "oversized-twiml": "split the flow across <Redirect> hops so each response "
                       "is small, and return an empty <Response/> to status "
                       "callbacks",
    "twiml-truncated": "the stored copy is truncated: generate the same document "
                       "locally and measure it in bytes",
    "not-twiml": "read the first line of the body and find what writes it",
    "no-body": "reproduce the request against the handler and measure what it "
               "writes",
}


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_alerts(session, since, limit, log_level="error"):
    """Page the Monitor alerts. next_page_url is absolute on this API."""
    url = MONITOR + "/Alerts"
    params = {"LogLevel": log_level, "StartDate": since, "PageSize": 1000}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("alerts", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def fetch_alert(session, sid):
    """One alert by SID: the only place response_body is populated."""
    return get(session, "%s/Alerts/%s" % (MONITOR, sid))


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=1,
                    help="how far back to read alerts (Twilio keeps 30 days)")
    ap.add_argument("--max-alerts", type=int, default=10000,
                    help="stop paging alerts after this many")
    ap.add_argument("--sample", type=int, default=1,
                    help="alerts to fetch individually per endpoint for the "
                         "response body (each one is a request)")
    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

    days = args.days
    if days > MAX_DAYS:
        log.warning("alerts are retained %d days; reading %d instead of %d",
                    MAX_DAYS, MAX_DAYS, days)
        days = MAX_DAYS

    session = requests.Session()
    session.auth = (key, secret)

    since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    alerts = list_alerts(session, since, args.max_alerts)
    rows = group(alerts)
    log.info("%d alert(s) since %s, %d endpoint(s) with 11750", len(alerts),
             since, len(rows))

    bad = 0
    for key, row in sorted(rows.items()):
        bad += 1
        state, detail = ("no-body", "not sampled")
        for sid in row["sids"][:max(1, args.sample)]:
            state, detail = classify_body(fetch_alert(session, sid)
                                          .get("response_body"))
            if state != "no-body":
                break
        log.warning("%-16s %s  %d x 11750  %s", state, key, row["alerts"], detail)
        log.warning("  first %s, last %s", row["first"], row["last"])
        log.warning("  repair: %s", REPAIRS.get(state, "read the body by hand"))

    log.info("%d endpoint(s) exceeding the %d byte TwiML limit", bad, LIMIT)
    return 1 if bad else 0


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

/**
 * Report Twilio webhooks whose response exceeds the 64 kB TwiML limit (11750).
 *
 * 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 MONITOR = 'https://monitor.twilio.com/v1';

const BODY_TOO_LARGE = 11750;

// The cap Twilio applies to a TwiML response, in bytes.
const LIMIT = 64 * 1024;

// Alerts are retained 30 days. A longer window is the same history mislabelled.
const MAX_DAYS = 30;

// Markers of a framework error page: the far commoner cause of 11750.
const TRACE_MARKERS = [
  'traceback (most recent call last)', 'stack trace', 'stacktrace',
  'whoops! there was an error', 'werkzeug debugger', 'actiondispatch',
];

/** Read error_code off an alert as a number, or null. */
export function codeOf(alert) {
  const raw = alert.error_code;
  if (raw === null || raw === undefined || raw === '') return null;
  const n = Number(raw);
  return Number.isFinite(n) ? n : null;
}

/** Lowercase host plus path, for grouping. Query string dropped. */
export function endpointOf(url) {
  if (!url) return '';
  const raw = String(url).trim();
  try {
    const u = new URL(raw);
    let path = u.pathname;
    while (path.endsWith('/')) path = path.slice(0, -1);
    return u.hostname.toLowerCase() + path;
  } catch {
    return raw.toLowerCase().replace(/\/+$/, '');
  }
}

/**
 * Size in bytes, which is the unit the limit is expressed in. A character count
 * reads a document with non-Latin text in <Say> as smaller than it is.
 */
export function byteLength(text) {
  return Buffer.byteLength(String(text ?? ''), 'utf8');
}

/**
 * Say what the oversized response actually was. Pure. Keys on what the body is
 * rather than how long it is, because response_body is stored truncated and its
 * length is a floor. Returns [state, detail].
 */
export function classifyBody(body) {
  const raw = body === null || body === undefined ? '' : String(body);
  const size = byteLength(raw);

  if (!raw.trim()) {
    return ['no-body',
      'the single-alert fetch returned nothing, so the cause cannot be read ' +
      'from here. Reproduce the request against the handler and measure what ' +
      'it writes.'];
  }

  const low = raw.replace(/^\s+/, '').toLowerCase();
  if (low.startsWith('<!doctype html') || low.startsWith('<html')
      || low.slice(0, 2000).includes('<html')) {
    return ['error-page',
      `an HTML page, not TwiML: at least ${size} bytes of framework debug ` +
      'output. The size is a symptom; the handler threw.'];
  }

  if (TRACE_MARKERS.some((m) => low.includes(m))) {
    return ['stack-trace',
      `a stack trace, at least ${size} bytes of it. Debug output is still on ` +
      'in production and every unhandled exception returns an essay.'];
  }

  if (low.includes('<response')) {
    if (size >= LIMIT) {
      return ['oversized-twiml',
        `real TwiML, ${size} bytes, over the ${LIMIT} byte cap. This one needs ` +
        'splitting rather than fixing.'];
    }
    return ['twiml-truncated',
      `real TwiML. The stored copy is ${size} bytes, under the cap, but ` +
      'response_body is truncated: that is a floor, not the size of the response.'];
  }

  return ['not-twiml',
    `at least ${size} bytes of something that is neither TwiML nor a ` +
    'recognisable error page. Read the first line of it.'];
}

/** Bucket alerts with one error code by endpoint. Pure. */
export function group(alerts, code = BODY_TOO_LARGE) {
  const out = new Map();
  for (const a of alerts) {
    if (codeOf(a) !== code) continue;
    const key = endpointOf(a.request_url);
    if (!out.has(key)) out.set(key, { alerts: 0, sids: [], first: null, last: null });
    const row = out.get(key);
    row.alerts += 1;
    if (row.sids.length < 3) row.sids.push(a.sid);
    const when = a.date_generated ?? '';
    if (when) {
      row.first = row.first === null || when < row.first ? when : row.first;
      row.last = row.last === null || when > row.last ? when : row.last;
    }
  }
  return out;
}

const REPAIRS = {
  'error-page': 'fix the exception, and turn debug pages off in production so a ' +
    'failure returns a short 500 rather than a rendered page',
  'stack-trace': 'disable debug output in production and return a small TwiML ' +
    'document from the error branch',
  'oversized-twiml': 'split the flow across <Redirect> hops so each response is ' +
    'small, and return an empty <Response/> to status callbacks',
  'twiml-truncated': 'the stored copy is truncated: generate the same document ' +
    'locally and measure it in bytes',
  'not-twiml': 'read the first line of the body and find what writes it',
  'no-body': 'reproduce the request against the handler and measure what it writes',
};

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 listAlerts(auth, since, limit = 10000, logLevel = 'error') {
  let url = `${MONITOR}/Alerts`;
  let params = { LogLevel: logLevel, StartDate: since, PageSize: 1000 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.alerts ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

/** One alert by SID: the only place response_body is populated. */
export async function fetchAlert(auth, sid) {
  return get(auth, `${MONITOR}/Alerts/${sid}`);
}

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

  let days = Number(process.argv.includes('--days')
    ? process.argv[process.argv.indexOf('--days') + 1] : 1) || 1;
  if (days > MAX_DAYS) {
    console.warn(`alerts are retained ${MAX_DAYS} days; reading ${MAX_DAYS} instead`);
    days = MAX_DAYS;
  }
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);

  const alerts = await listAlerts(auth, since);
  const rows = group(alerts);
  console.log(`${alerts.length} alert(s) since ${since}, ${rows.size} endpoint(s) with 11750`);

  let bad = 0;
  for (const [key, row] of [...rows.entries()].sort()) {
    bad += 1;
    let state = 'no-body';
    let detail = 'not sampled';
    for (const sid of row.sids.slice(0, 1)) {
      const full = await fetchAlert(auth, sid);
      [state, detail] = classifyBody(full.response_body);
    }
    console.warn(`${state.padEnd(16)} ${key}  ${row.alerts} x 11750  ${detail}`);
    console.warn(`  first ${row.first}, last ${row.last}`);
    console.warn(`  repair: ${REPAIRS[state] ?? 'read the body by hand'}`);
  }

  console.log(`${bad} endpoint(s) exceeding the ${LIMIT} byte TwiML limit`);
  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_twiml_size_audit import (byte_length, classify_body, code_of,
                                       endpoint_of, group, LIMIT)


def alert(sid, url, code="11750", when="2026-04-02T10:00:00Z"):
    return {"sid": sid, "request_url": url, "error_code": code,
            "date_generated": when}


def test_the_limit_is_measured_in_bytes_not_characters():
    # The mistake this exists to prevent: len() on a str reads a failing
    # document as comfortably inside the cap.
    assert byte_length("caf\u00e9") == 5
    assert len("caf\u00e9") == 4
    assert byte_length("\U0001f600") == 4
    assert byte_length(None) == 0


def test_a_framework_debug_page_is_the_usual_cause():
    state, detail = classify_body("<!DOCTYPE html><html><body>Server Error</body></html>")
    assert state == "error-page"
    assert "symptom" in detail


def test_a_stack_trace_is_named_separately_from_a_rendered_page():
    state, _ = classify_body("Traceback (most recent call last):\n  File ...")
    assert state == "stack-trace"


def test_genuine_twiml_over_the_cap_is_a_splitting_problem():
    body = "<Response>" + ("<Say>hello</Say>" * 6000) + "</Response>"
    assert byte_length(body) > LIMIT
    state, detail = classify_body(body)
    assert state == "oversized-twiml"
    assert "splitting" in detail


def test_twiml_under_the_cap_is_reported_as_a_floor_not_a_clean_bill():
    # response_body is stored truncated, so a small stored copy proves nothing
    # about the response Twilio actually refused.
    state, detail = classify_body("<Response><Say>Hi</Say></Response>")
    assert state == "twiml-truncated"
    assert "floor" in detail


def test_an_empty_body_is_reported_rather_than_guessed():
    assert classify_body("")[0] == "no-body"
    assert classify_body(None)[0] == "no-body"


def test_something_that_is_neither_twiml_nor_an_error_page():
    state, detail = classify_body('{"error": "too many participants"}')
    assert state == "not-twiml"
    assert "bytes" in detail


def test_group_keeps_only_11750_and_records_the_ends():
    rows = group([alert("NO1", "https://a.example.com/voice?CallSid=CA1",
                        when="2026-04-02T10:00:00Z"),
                  alert("NO2", "https://a.example.com/voice/",
                        when="2026-04-01T09:00:00Z"),
                  alert("NO3", "https://a.example.com/voice", code="12100")])
    assert set(rows) == {"a.example.com/voice"}
    assert rows["a.example.com/voice"]["alerts"] == 2
    assert rows["a.example.com/voice"]["last"] == "2026-04-02T10:00:00Z"


def test_code_and_endpoint_helpers():
    assert code_of({"error_code": "11750"}) == 11750
    assert code_of({}) is None
    assert endpoint_of("https://A.example.com/voice?x=1") == "a.example.com/voice"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  byteLength, classifyBody, codeOf, endpointOf, group,
} from './twilio-twiml-size-audit.mjs';

const LIMIT = 64 * 1024;

const alert = (sid, url, code = '11750', when = '2026-04-02T10:00:00Z') => ({
  sid, request_url: url, error_code: code, date_generated: when,
});

test('the limit is measured in bytes, not characters', () => {
  assert.equal(byteLength('caf\u00e9'), 5);
  assert.equal('caf\u00e9'.length, 4);
  assert.equal(byteLength('\u{1f600}'), 4);
  assert.equal(byteLength(null), 0);
});

test('a framework debug page is the usual cause', () => {
  const [state, detail] = classifyBody('<!DOCTYPE html><html><body>Server Error</body></html>');
  assert.equal(state, 'error-page');
  assert.match(detail, /symptom/);
});

test('a stack trace is named separately from a rendered page', () => {
  assert.equal(classifyBody('Traceback (most recent call last):\n  File ...')[0],
    'stack-trace');
});

test('genuine TwiML over the cap is a splitting problem', () => {
  const body = `<Response>${'<Say>hello</Say>'.repeat(6000)}</Response>`;
  assert.ok(byteLength(body) > LIMIT);
  const [state, detail] = classifyBody(body);
  assert.equal(state, 'oversized-twiml');
  assert.match(detail, /splitting/);
});

test('TwiML under the cap is reported as a floor, not a clean bill', () => {
  const [state, detail] = classifyBody('<Response><Say>Hi</Say></Response>');
  assert.equal(state, 'twiml-truncated');
  assert.match(detail, /floor/);
});

test('an empty body is reported rather than guessed', () => {
  assert.equal(classifyBody('')[0], 'no-body');
  assert.equal(classifyBody(null)[0], 'no-body');
});

test('something that is neither TwiML nor an error page', () => {
  const [state, detail] = classifyBody('{"error": "too many participants"}');
  assert.equal(state, 'not-twiml');
  assert.match(detail, /bytes/);
});

test('group keeps only 11750 and records the ends', () => {
  const rows = group([
    alert('NO1', 'https://a.example.com/voice?CallSid=CA1', '11750', '2026-04-02T10:00:00Z'),
    alert('NO2', 'https://a.example.com/voice/', '11750', '2026-04-01T09:00:00Z'),
    alert('NO3', 'https://a.example.com/voice', '12100'),
  ]);
  assert.deepEqual([...rows.keys()], ['a.example.com/voice']);
  assert.equal(rows.get('a.example.com/voice').alerts, 2);
  assert.equal(rows.get('a.example.com/voice').last, '2026-04-02T10:00:00Z');
});

test('code and endpoint helpers', () => {
  assert.equal(codeOf({ error_code: '11750' }), 11750);
  assert.equal(codeOf({}), null);
  assert.equal(endpointOf('https://A.example.com/voice?x=1'), 'a.example.com/voice');
});