a TwiML response with the wrong Content-Type fails with 12300

"""Report Twilio webhooks returning a Content-Type that TwiML parsing rejects.

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

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

INVALID_CONTENT_TYPE = 12300

# 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

# The two media types Twilio will parse as TwiML.
TWIML_TYPES = ("text/xml", "application/xml")


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 header_value(headers, name):
    """Case-insensitively read one header out of an alert's response_headers.

    Two things make this less trivial than it looks. Header names are not
    case-sensitive, so a lookup for "Content-Type" has to find "content-type".
    And the field does not arrive in one shape: it can be a mapping, a list of
    lines, or a single blob using either ':' or '=' between name and value.
    Supporting only the shape you saw first is how a check stops matching.
    """
    want = str(name).strip().lower()
    if headers is None:
        return ""
    if isinstance(headers, dict):
        for k, v in headers.items():
            if str(k).strip().lower() == want:
                return str(v).strip()
        return ""
    if isinstance(headers, (list, tuple)):
        lines = [str(h) for h in headers]
    else:
        lines = str(headers).replace("\r\n", "\n").replace("&", "\n").split("\n")
    for line in lines:
        for sep in (":", "="):
            k, found, v = line.partition(sep)
            if found and k.strip().lower() == want:
                return v.strip()
    return ""


def media_type(value):
    """The media type with its parameters stripped.

    'text/xml; charset=utf-8' is a correct TwiML response and an exact-match
    check on 'text/xml' rejects it. Everything after the semicolon is a
    parameter and none of it changes how Twilio routes the response.
    """
    return str(value or "").split(";", 1)[0].strip().lower()


def content_type_verdict(value):
    """Classify one Content-Type. Pure, so every branch is testable offline.

    Returns (state, detail).
    """
    mt = media_type(value)

    if not mt:
        return ("missing",
                "no Content-Type at all. Twilio has nothing to dispatch on, and "
                "the Debugger shows this as 502 Bad Gateway rather than 12300, "
                "which is why it gets chased as a gateway problem.")

    if mt in TWIML_TYPES:
        return ("ok", "%s is parsed as TwiML" % mt)

    if mt.startswith("audio/"):
        return ("audio",
                "%s is an audio type, so this alert is about a <Play> target "
                "rather than your TwiML. Fix the file that URL serves, not the "
                "webhook." % mt)

    if mt in ("text/html", "application/xhtml+xml"):
        return ("html",
                "%s is the framework default when nothing sets the header. The "
                "body may be perfect TwiML; Twilio never reads it." % mt)

    if mt in ("application/json", "text/json"):
        return ("json",
                "%s means an API handler is answering a TwiML webhook. Either "
                "the route is wrong or the serialiser is." % mt)

    if mt == "text/plain":
        return ("plain",
                "text/plain is what a bare string return produces. Set the "
                "header explicitly on every branch of the handler.")

    if mt.endswith("+xml"):
        return ("odd-xml",
                "%s is XML-shaped but is not one of the two media types Twilio "
                "dispatches TwiML on. Send text/xml or application/xml." % mt)

    return ("other",
            "%s is not a media type Twilio parses as TwiML." % mt)


def group(alerts, code=INVALID_CONTENT_TYPE):
    """Bucket alerts with one error code by endpoint. Pure.

    date_generated is ISO 8601 in UTC, so a string comparison finds the ends of
    the window without parsing anything.
    """
    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


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_headers 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 headers (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 12300", len(alerts),
             since, len(rows))

    bad = 0
    for key, row in sorted(rows.items()):
        sent = ""
        for sid in row["sids"][:max(1, args.sample)]:
            sent = header_value(fetch_alert(session, sid).get("response_headers"),
                                "Content-Type")
            if sent:
                break
        state, detail = content_type_verdict(sent)
        log.warning("%-8s %s  %d x 12300  %s", state, key, row["alerts"], detail)
        log.warning("  first %s, last %s", row["first"], row["last"])
        if state == "ok":
            log.warning("  the sampled alert carried a valid TwiML type: sample "
                        "more alerts, the failing responses came from another "
                        "branch of the handler")
            continue
        bad += 1
        if state == "audio":
            log.warning("  repair: serve that <Play> URL as audio/mpeg or "
                        "audio/wav; it is currently an HTML or error response")
        else:
            log.warning("  repair: set Content-Type: text/xml on this response, "
                        "on every branch of the handler including the error "
                        "branches")

    log.info("%d endpoint(s) returning a Content-Type Twilio will not parse", bad)
    return 1 if bad else 0


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

/**
 * Report Twilio webhooks returning a Content-Type that TwiML parsing rejects.
 *
 * 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 INVALID_CONTENT_TYPE = 12300;

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

// The two media types Twilio will parse as TwiML.
const TWIML_TYPES = ['text/xml', 'application/xml'];

/** 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(/\/+$/, '');
  }
}

/**
 * Case-insensitively read one header out of an alert's response_headers. The
 * field arrives as a mapping, a list of lines or one blob using ':' or '='.
 */
export function headerValue(headers, name) {
  const want = String(name).trim().toLowerCase();
  if (headers === null || headers === undefined) return '';
  if (!Array.isArray(headers) && typeof headers === 'object') {
    for (const [k, v] of Object.entries(headers)) {
      if (k.trim().toLowerCase() === want) return String(v).trim();
    }
    return '';
  }
  const lines = Array.isArray(headers)
    ? headers.map(String)
    : String(headers).replace(/\r\n/g, '\n').split(/[\n&]/);
  for (const line of lines) {
    for (const sep of [':', '=']) {
      const at = line.indexOf(sep);
      if (at > 0 && line.slice(0, at).trim().toLowerCase() === want) {
        return line.slice(at + 1).trim();
      }
    }
  }
  return '';
}

/**
 * The media type with its parameters stripped. 'text/xml; charset=utf-8' is a
 * correct TwiML response and an exact-match check rejects it.
 */
export function mediaType(value) {
  return String(value ?? '').split(';')[0].trim().toLowerCase();
}

/** Classify one Content-Type. Pure. Returns [state, detail]. */
export function contentTypeVerdict(value) {
  const mt = mediaType(value);

  if (!mt) {
    return ['missing',
      'no Content-Type at all. Twilio has nothing to dispatch on, and the ' +
      'Debugger shows this as 502 Bad Gateway rather than 12300, which is why ' +
      'it gets chased as a gateway problem.'];
  }

  if (TWIML_TYPES.includes(mt)) return ['ok', `${mt} is parsed as TwiML`];

  if (mt.startsWith('audio/')) {
    return ['audio',
      `${mt} is an audio type, so this alert is about a <Play> target rather ` +
      'than your TwiML. Fix the file that URL serves, not the webhook.'];
  }

  if (mt === 'text/html' || mt === 'application/xhtml+xml') {
    return ['html',
      `${mt} is the framework default when nothing sets the header. The body ` +
      'may be perfect TwiML; Twilio never reads it.'];
  }

  if (mt === 'application/json' || mt === 'text/json') {
    return ['json',
      `${mt} means an API handler is answering a TwiML webhook. Either the ` +
      'route is wrong or the serialiser is.'];
  }

  if (mt === 'text/plain') {
    return ['plain',
      'text/plain is what a bare string return produces. Set the header ' +
      'explicitly on every branch of the handler.'];
  }

  if (mt.endsWith('+xml')) {
    return ['odd-xml',
      `${mt} is XML-shaped but is not one of the two media types Twilio ` +
      'dispatches TwiML on. Send text/xml or application/xml.'];
  }

  return ['other', `${mt} is not a media type Twilio parses as TwiML.`];
}

/** Bucket alerts with one error code by endpoint. Pure. */
export function group(alerts, code = INVALID_CONTENT_TYPE) {
  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;
}

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_headers 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 12300`);

  let bad = 0;
  for (const [key, row] of [...rows.entries()].sort()) {
    let sent = '';
    for (const sid of row.sids.slice(0, 1)) {
      const full = await fetchAlert(auth, sid);
      sent = headerValue(full.response_headers, 'Content-Type');
    }
    const [state, detail] = contentTypeVerdict(sent);
    console.warn(`${state.padEnd(8)} ${key}  ${row.alerts} x 12300  ${detail}`);
    console.warn(`  first ${row.first}, last ${row.last}`);
    if (state === 'ok') {
      console.warn('  the sampled alert carried a valid TwiML type: sample more ' +
                   'alerts, the failing responses came from another branch of ' +
                   'the handler');
      continue;
    }
    bad += 1;
    if (state === 'audio') {
      console.warn('  repair: serve that <Play> URL as audio/mpeg or audio/wav; ' +
                   'it is currently an HTML or error response');
    } else {
      console.warn('  repair: set Content-Type: text/xml on this response, on ' +
                   'every branch of the handler including the error branches');
    }
  }

  console.log(`${bad} endpoint(s) returning a Content-Type Twilio will not parse`);
  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_content_type_audit import (code_of, content_type_verdict, endpoint_of,
                                         group, header_value, media_type)


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


def test_code_of_reads_the_string_the_monitor_api_returns():
    assert code_of({"error_code": "12300"}) == 12300
    assert code_of({"error_code": 12300}) == 12300
    assert code_of({}) is None


def test_a_charset_parameter_does_not_make_the_type_wrong():
    # The single most common false positive: this response is correct.
    assert content_type_verdict("text/xml; charset=utf-8")[0] == "ok"
    assert content_type_verdict("TEXT/XML")[0] == "ok"
    assert content_type_verdict("application/xml")[0] == "ok"


def test_a_missing_header_is_its_own_state_because_it_reads_as_502():
    state, detail = content_type_verdict("")
    assert state == "missing"
    assert "502" in detail
    assert content_type_verdict(None)[0] == "missing"


def test_html_json_and_plain_are_told_apart():
    assert content_type_verdict("text/html; charset=utf-8")[0] == "html"
    assert content_type_verdict("application/json")[0] == "json"
    assert content_type_verdict("text/plain")[0] == "plain"


def test_an_audio_type_means_the_alert_is_about_a_play_target():
    state, detail = content_type_verdict("audio/mpeg")
    assert state == "audio"
    assert "<Play>" in detail


def test_an_xml_flavoured_type_is_still_not_twiml():
    assert content_type_verdict("application/soap+xml")[0] == "odd-xml"
    assert content_type_verdict("application/pdf")[0] == "other"


def test_header_lookup_is_case_insensitive_across_every_shape():
    assert header_value({"content-type": "text/html"}, "Content-Type") == "text/html"
    assert header_value(["Server: nginx", "Content-Type: text/html"],
                        "Content-Type") == "text/html"
    assert header_value("Server: nginx\nContent-Type: text/html",
                        "content-type") == "text/html"
    assert header_value("Server=nginx&Content-Type=application/json",
                        "Content-Type") == "application/json"
    assert header_value(None, "Content-Type") == ""


def test_media_type_strips_parameters_and_whitespace():
    assert media_type("  Text/XML ; charset=UTF-8 ") == "text/xml"
    assert media_type(None) == ""


def test_group_keeps_only_the_requested_code_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"]["first"] == "2026-04-01T09:00:00Z"


def test_endpoint_of_drops_the_query_string_twilio_appends():
    assert endpoint_of("https://A.example.com/voice?CallSid=CA1") == \
        "a.example.com/voice"
    assert endpoint_of(None) == ""

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  codeOf, contentTypeVerdict, endpointOf, group, headerValue, mediaType,
} from './twilio-content-type-audit.mjs';

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

test('codeOf reads the string the Monitor API returns', () => {
  assert.equal(codeOf({ error_code: '12300' }), 12300);
  assert.equal(codeOf({ error_code: 12300 }), 12300);
  assert.equal(codeOf({}), null);
});

test('a charset parameter does not make the type wrong', () => {
  assert.equal(contentTypeVerdict('text/xml; charset=utf-8')[0], 'ok');
  assert.equal(contentTypeVerdict('TEXT/XML')[0], 'ok');
  assert.equal(contentTypeVerdict('application/xml')[0], 'ok');
});

test('a missing header is its own state because it reads as 502', () => {
  const [state, detail] = contentTypeVerdict('');
  assert.equal(state, 'missing');
  assert.match(detail, /502/);
  assert.equal(contentTypeVerdict(null)[0], 'missing');
});

test('html, json and plain are told apart', () => {
  assert.equal(contentTypeVerdict('text/html; charset=utf-8')[0], 'html');
  assert.equal(contentTypeVerdict('application/json')[0], 'json');
  assert.equal(contentTypeVerdict('text/plain')[0], 'plain');
});

test('an audio type means the alert is about a Play target', () => {
  const [state, detail] = contentTypeVerdict('audio/mpeg');
  assert.equal(state, 'audio');
  assert.match(detail, /<Play>/);
});

test('an xml-flavoured type is still not TwiML', () => {
  assert.equal(contentTypeVerdict('application/soap+xml')[0], 'odd-xml');
  assert.equal(contentTypeVerdict('application/pdf')[0], 'other');
});

test('header lookup is case-insensitive across every shape', () => {
  assert.equal(headerValue({ 'content-type': 'text/html' }, 'Content-Type'), 'text/html');
  assert.equal(headerValue(['Server: nginx', 'Content-Type: text/html'], 'Content-Type'),
    'text/html');
  assert.equal(headerValue('Server: nginx\nContent-Type: text/html', 'content-type'),
    'text/html');
  assert.equal(headerValue('Server=nginx&Content-Type=application/json', 'Content-Type'),
    'application/json');
  assert.equal(headerValue(null, 'Content-Type'), '');
});

test('mediaType strips parameters and whitespace', () => {
  assert.equal(mediaType('  Text/XML ; charset=UTF-8 '), 'text/xml');
  assert.equal(mediaType(null), '');
});

test('group keeps only the requested code and records the ends', () => {
  const rows = group([
    alert('NO1', 'https://a.example.com/voice?CallSid=CA1', '12300', '2026-04-02T10:00:00Z'),
    alert('NO2', 'https://a.example.com/voice/', '12300', '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').first, '2026-04-01T09:00:00Z');
});

test('endpointOf drops the query string Twilio appends', () => {
  assert.equal(endpointOf('https://A.example.com/voice?CallSid=CA1'), 'a.example.com/voice');
  assert.equal(endpointOf(null), '');
});