12200: TwiML that parses, fails the schema, and is skipped

"""Report TwiML that parses and then fails the schema: error 12200.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The repair is in your own template, and it
is printed rather than performed.
"""
import argparse
import datetime as dt
import logging
import os
import re
import sys

import requests

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

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

VERBS = {
    "Response", "Say", "Play", "Gather", "Record", "Dial", "Sms", "Message",
    "Body", "Media", "Redirect", "Hangup", "Reject", "Pause", "Enqueue", "Leave",
    "Queue", "Conference", "Number", "Client", "Sip", "Task", "Refer", "Pay",
    "Prompt", "Parameter", "Connect", "Stream", "Start", "Stop", "Siprec",
    "VirtualAgent", "Identity", "Room", "Application",
}
VERB_BY_LOWER = {v.lower(): v for v in VERBS}

# Only the camelCase attributes: those are where the casing mistakes happen, and
# limiting the list to them keeps the scanner from inventing findings about
# attributes it simply has not heard of.
ATTRS = [
    "numDigits", "finishOnKey", "speechTimeout", "speechModel", "actionOnEmptyResult",
    "partialResultCallback", "partialResultCallbackMethod", "callerId", "timeLimit",
    "hangupOnStar", "answerOnBridge", "ringTone", "recordingStatusCallback",
    "recordingStatusCallbackMethod", "recordingStatusCallbackEvent", "maxLength",
    "playBeep", "transcribeCallback", "statusCallback", "statusCallbackEvent",
    "statusCallbackMethod", "waitUrl", "waitMethod", "startConferenceOnEnter",
    "endConferenceOnExit", "maxParticipants", "sendDigits", "machineDetection",
    "referUrl", "maxSpeechTime", "profanityFilter", "playTone", "recordingTrack",
]
ATTR_BY_LOWER = {a.lower(): a for a in ATTRS}

TAG = re.compile(r"<\s*(/?)\s*([A-Za-z_][A-Za-z0-9_.-]*)([^<>]*?)/?>", re.S)
ATTR_NAME = re.compile(r"([A-Za-z_][A-Za-z0-9_.:-]*)\s*=")
SAY_BLOCK = re.compile(r"(<\s*[Ss][Aa][Yy]\b[^<>]*>)(.*?)(<\s*/\s*[Ss][Aa][Yy]\s*>)", re.S)


def code_of(alert):
    """error_code arrives as a string on some alerts and an int on others."""
    try:
        return int(alert.get("error_code"))
    except (TypeError, ValueError):
        return None


def strip_say_children(xml):
    """Drop what is inside <Say>, keeping the tags themselves.

    Pure. SSML is lower-case by design: <break>, <prosody> and <say-as> are
    correct exactly as written, and a scanner that flags them reports a healthy
    document as broken. The Say tags stay so their own casing is still checked.
    """
    return SAY_BLOCK.sub(lambda m: m.group(1) + m.group(3), str(xml or ""))


def scan(xml):
    """Find the schema mistakes in a TwiML document. Pure, so the vocabulary
    rules can be tested without a network.

    Returns a list of (kind, found, suggestion). This is not a validator: it is
    a check for the two mistakes that produce almost every 12200.
    """
    body = strip_say_children(xml)
    findings, seen, root_checked = [], set(), False

    def note(kind, found, suggestion):
        keyps = (kind, found)
        if keyps in seen:
            return
        seen.add(keyps)
        findings.append((kind, found, suggestion))

    for match in TAG.finditer(body):
        closing, name, rest = match.group(1), match.group(2), match.group(3) or ""

        if not root_checked and not closing:
            root_checked = True
            if name != "Response":
                if name.lower() == "response":
                    note("verb-casing", name, "Response")
                else:
                    note("root", name, "Response")
                continue

        if name not in VERBS:
            canonical = VERB_BY_LOWER.get(name.lower())
            if canonical:
                note("verb-casing", name, canonical)
            else:
                note("unknown-verb", name, None)
            continue

        if closing:
            continue
        for attr in ATTR_NAME.findall(rest):
            if attr in ATTR_BY_LOWER.values():
                continue
            canonical = ATTR_BY_LOWER.get(attr.lower())
            if canonical:
                note("attribute-casing", attr, canonical)

    return findings


def verdict(findings, count=1):
    """Turn the scan into one line for the report. Pure. Returns (state, detail)."""
    by_kind = {}
    for kind, found, suggestion in findings:
        by_kind.setdefault(kind, []).append((found, suggestion))

    def named(kind):
        return ", ".join("%s should be %s" % (f, s) if s else f
                         for f, s in by_kind[kind][:4])

    if "verb-casing" in by_kind:
        return ("verb-casing",
                "%d alert(s): %s. TwiML is case-sensitive, so the verb is "
                "skipped and the call continues past it." % (count, named("verb-casing")))
    if "attribute-casing" in by_kind:
        return ("attribute-casing",
                "%d alert(s): %s. The attribute is dropped and the verb runs on "
                "its default." % (count, named("attribute-casing")))
    if "root" in by_kind:
        return ("bad-root",
                "%d alert(s): the document root is %s. Every TwiML document has "
                "to be <Response>." % (count, named("root")))
    if "unknown-verb" in by_kind:
        return ("unknown-verb",
                "%d alert(s): %s is not in the TwiML vocabulary at all, so it is "
                "not a casing slip." % (count, named("unknown-verb")))
    return ("unexplained",
            "%d alert(s) and the scanner found no casing or vocabulary mistake: "
            "read alert_text for the line and column, and check the nesting."
            % count)


def endpoint(url):
    """Host plus path, lowercased. One bad template fires on every call through
    it, and grouping on the raw URL hides that."""
    u = str(url or "").strip()
    for scheme in ("https://", "http://"):
        if u.lower().startswith(scheme):
            u = u[len(scheme):]
            break
    return u.split("?", 1)[0].split("#", 1)[0].rstrip("/").lower()


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=10000, log_level="warning"):
    url = MONITOR + "/Alerts"
    params = {"PageSize": 100, "LogLevel": log_level, "StartDate": since}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("alerts", []))
        url, params = (page.get("meta") or {}).get("next_page_url"), {}
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=7, help="alert window, max 30")
    ap.add_argument("--sample", action="store_true",
                    help="one extra GET per endpoint to read the actual document")
    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)

    since = (dt.datetime.now(dt.timezone.utc)
             - dt.timedelta(days=min(args.days, 30))).strftime("%Y-%m-%dT%H:%M:%SZ")
    # LogLevel=warning is the point of this script: 12200 never appears in an
    # error-only sweep, which is why accounts carry it for months.
    alerts = [a for a in list_alerts(session, since) if code_of(a) == 12200]
    if not alerts:
        log.info("0 endpoint(s) emitting 12200 in the last %d day(s)", args.days)
        return 0

    rows = {}
    for a in alerts:
        e = endpoint(a.get("request_url"))
        row = rows.setdefault(e, {"count": 0, "sid": a.get("sid"),
                                  "text": a.get("alert_text") or ""})
        row["count"] += 1

    bad = 0
    for e, row in sorted(rows.items(), key=lambda kv: -kv[1]["count"]):
        findings = []
        if args.sample and row["sid"]:
            full = get(session, "%s/Alerts/%s" % (MONITOR, row["sid"]))
            findings = scan(full.get("response_body"))
        state, detail = verdict(findings, row["count"])
        bad += 1
        log.warning("%-18s %s  %s", state, e or "unknown endpoint", detail)
        if not args.sample:
            log.warning("  re-run with --sample to read the document Twilio received")
        log.warning("  repair: correct the casing in the template that renders "
                    "this document; alert_text gives the line and column: %s",
                    row["text"][:160])

    log.info("%d endpoint(s) emitting 12200 in the last %d day(s)", bad, args.days)
    return 1 if bad else 0


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

/**
 * Report TwiML that parses and then fails the schema: error 12200.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The repair is in your own
 * template, and it is printed rather than performed.
 */
const MONITOR = 'https://monitor.twilio.com/v1';

const VERBS = new Set([
  'Response', 'Say', 'Play', 'Gather', 'Record', 'Dial', 'Sms', 'Message',
  'Body', 'Media', 'Redirect', 'Hangup', 'Reject', 'Pause', 'Enqueue', 'Leave',
  'Queue', 'Conference', 'Number', 'Client', 'Sip', 'Task', 'Refer', 'Pay',
  'Prompt', 'Parameter', 'Connect', 'Stream', 'Start', 'Stop', 'Siprec',
  'VirtualAgent', 'Identity', 'Room', 'Application',
]);
const VERB_BY_LOWER = new Map([...VERBS].map((v) => [v.toLowerCase(), v]));

// Only the camelCase attributes: those are where the casing mistakes happen,
// and limiting the list to them keeps the scanner from inventing findings about
// attributes it simply has not heard of.
const ATTRS = [
  'numDigits', 'finishOnKey', 'speechTimeout', 'speechModel', 'actionOnEmptyResult',
  'partialResultCallback', 'partialResultCallbackMethod', 'callerId', 'timeLimit',
  'hangupOnStar', 'answerOnBridge', 'ringTone', 'recordingStatusCallback',
  'recordingStatusCallbackMethod', 'recordingStatusCallbackEvent', 'maxLength',
  'playBeep', 'transcribeCallback', 'statusCallback', 'statusCallbackEvent',
  'statusCallbackMethod', 'waitUrl', 'waitMethod', 'startConferenceOnEnter',
  'endConferenceOnExit', 'maxParticipants', 'sendDigits', 'machineDetection',
  'referUrl', 'maxSpeechTime', 'profanityFilter', 'playTone', 'recordingTrack',
];
const ATTR_BY_LOWER = new Map(ATTRS.map((a) => [a.toLowerCase(), a]));
const ATTR_SET = new Set(ATTRS);

const TAG = /<\s*(\/?)\s*([A-Za-z_][A-Za-z0-9_.-]*)([^<>]*?)\/?>/gs;
const ATTR_NAME = /([A-Za-z_][A-Za-z0-9_.:-]*)\s*=/g;
const SAY_BLOCK = /(<\s*[Ss][Aa][Yy]\b[^<>]*>)([\s\S]*?)(<\s*\/\s*[Ss][Aa][Yy]\s*>)/g;

/** error_code arrives as a string on some alerts and a number on others. */
export function codeOf(alert) {
  const n = Number(alert?.error_code);
  return Number.isFinite(n) ? n : null;
}

/**
 * Drop what is inside <Say>, keeping the tags themselves. Pure. SSML is
 * lower-case by design, and a scanner that flags it reports a healthy document
 * as broken. The Say tags stay so their own casing is still checked.
 */
export function stripSayChildren(xml) {
  return String(xml ?? '').replace(SAY_BLOCK, (m, open, inner, close) => open + close);
}

/**
 * Find the schema mistakes in a TwiML document. Pure, so the vocabulary rules
 * can be tested without a network. Returns [kind, found, suggestion] triples.
 * This is not a validator: it is a check for the two mistakes that produce
 * almost every 12200.
 */
export function scan(xml) {
  const body = stripSayChildren(xml);
  const findings = [];
  const seen = new Set();
  let rootChecked = false;

  const note = (kind, found, suggestion) => {
    const k = `${kind}\u0000${found}`;
    if (seen.has(k)) return;
    seen.add(k);
    findings.push([kind, found, suggestion]);
  };

  for (const match of body.matchAll(TAG)) {
    const closing = match[1];
    const name = match[2];
    const rest = match[3] ?? '';

    if (!rootChecked && !closing) {
      rootChecked = true;
      if (name !== 'Response') {
        if (name.toLowerCase() === 'response') note('verb-casing', name, 'Response');
        else note('root', name, 'Response');
        continue;
      }
    }

    if (!VERBS.has(name)) {
      const canonical = VERB_BY_LOWER.get(name.toLowerCase());
      if (canonical) note('verb-casing', name, canonical);
      else note('unknown-verb', name, null);
      continue;
    }

    if (closing) continue;
    for (const attr of rest.matchAll(ATTR_NAME)) {
      const found = attr[1];
      if (ATTR_SET.has(found)) continue;
      const canonical = ATTR_BY_LOWER.get(found.toLowerCase());
      if (canonical) note('attribute-casing', found, canonical);
    }
  }

  return findings;
}

/** Turn the scan into one line for the report. Pure. Returns [state, detail]. */
export function verdict(findings, count = 1) {
  const byKind = new Map();
  for (const [kind, found, suggestion] of findings) {
    if (!byKind.has(kind)) byKind.set(kind, []);
    byKind.get(kind).push([found, suggestion]);
  }
  const named = (kind) => byKind.get(kind).slice(0, 4)
    .map(([f, s]) => (s ? `${f} should be ${s}` : f)).join(', ');

  if (byKind.has('verb-casing')) {
    return ['verb-casing',
      `${count} alert(s): ${named('verb-casing')}. TwiML is case-sensitive, so ` +
      'the verb is skipped and the call continues past it.'];
  }
  if (byKind.has('attribute-casing')) {
    return ['attribute-casing',
      `${count} alert(s): ${named('attribute-casing')}. The attribute is ` +
      'dropped and the verb runs on its default.'];
  }
  if (byKind.has('root')) {
    return ['bad-root',
      `${count} alert(s): the document root is ${named('root')}. Every TwiML ` +
      'document has to be <Response>.'];
  }
  if (byKind.has('unknown-verb')) {
    return ['unknown-verb',
      `${count} alert(s): ${named('unknown-verb')} is not in the TwiML ` +
      'vocabulary at all, so it is not a casing slip.'];
  }
  return ['unexplained',
    `${count} alert(s) and the scanner found no casing or vocabulary mistake: ` +
    'read alert_text for the line and column, and check the nesting.'];
}

/** Host plus path, lowercased: one bad template fires on every call through it. */
export function endpoint(url) {
  let u = String(url ?? '').trim();
  for (const scheme of ['https://', 'http://']) {
    if (u.toLowerCase().startsWith(scheme)) { u = u.slice(scheme.length); break; }
  }
  return u.split('?')[0].split('#')[0].replace(/\/+$/, '').toLowerCase();
}

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

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 days = Math.min(Number(process.env.DAYS ?? 7), 30);
  const sample = process.argv.includes('--sample');

  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 19) + 'Z';
  // LogLevel=warning is the point of this script: 12200 never appears in an
  // error-only sweep, which is why accounts carry it for months.
  const alerts = (await listAlerts(auth, since)).filter((a) => codeOf(a) === 12200);
  if (alerts.length === 0) {
    console.log(`0 endpoint(s) emitting 12200 in the last ${days} day(s)`);
    return;
  }

  const rows = new Map();
  for (const a of alerts) {
    const e = endpoint(a.request_url);
    const row = rows.get(e) ?? { count: 0, sid: a.sid, text: a.alert_text ?? '' };
    row.count += 1;
    rows.set(e, row);
  }

  let bad = 0;
  for (const [e, row] of [...rows.entries()].sort((a, b) => b[1].count - a[1].count)) {
    let findings = [];
    if (sample && row.sid) {
      const full = await get(auth, `${MONITOR}/Alerts/${row.sid}`);
      findings = scan(full.response_body);
    }
    const [state, detail] = verdict(findings, row.count);
    bad += 1;
    console.warn(`${state.padEnd(18)} ${e || 'unknown endpoint'}  ${detail}`);
    if (!sample) console.warn('  re-run with --sample to read the document Twilio received');
    console.warn('  repair: correct the casing in the template that renders this ' +
                 `document; alert_text gives the line and column: ${row.text.slice(0, 160)}`);
  }

  console.log(`${bad} endpoint(s) emitting 12200 in the last ${days} day(s)`);
  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_schema_audit import scan, strip_say_children, verdict

GOOD = """<?xml version="1.0" encoding="UTF-8"?>
<Response><Gather numDigits="4" action="/entered"><Say>Enter your code</Say></Gather></Response>"""


def test_a_correct_document_produces_nothing():
    assert scan(GOOD) == []
    assert verdict(scan(GOOD), 0)[0] == "unexplained"


def test_ssml_inside_say_is_not_a_casing_error():
    # The false positive that would get this script switched off.
    doc = '<Response><Say>One<break time="500ms"/>two<say-as>3</say-as></Say></Response>'
    assert scan(doc) == []


def test_a_lowercase_say_is_still_caught_even_though_say_is_exempt():
    doc = "<Response><say>hello<break/></say></Response>"
    findings = scan(doc)
    assert ("verb-casing", "say", "Say") in findings
    assert not any(f[1] == "break" for f in findings)


def test_a_miscased_attribute_names_the_camelcase_form():
    doc = '<Response><Gather numdigits="4"/></Response>'
    state, detail = verdict(scan(doc), 12)
    assert state == "attribute-casing"
    assert "numdigits should be numDigits" in detail
    assert "12 alert(s)" in detail


def test_an_unknown_verb_is_not_reported_as_a_casing_slip():
    state, detail = verdict(scan("<Response><Speak>hi</Speak></Response>"))
    assert state == "unknown-verb"
    assert "not in the TwiML vocabulary" in detail


def test_a_root_that_is_not_response_is_its_own_state():
    state, detail = verdict(scan("<Twiml><Say>hi</Say></Twiml>"))
    assert state == "bad-root"
    assert "<Response>" in detail


def test_a_lowercase_root_is_a_casing_finding_not_a_bad_root():
    assert ("verb-casing", "response", "Response") in scan("<response><Say>hi</Say></response>")


def test_strip_say_children_keeps_the_tags_it_removes_the_contents_of():
    out = strip_say_children("<Response><Say voice=\"alice\"><break/></Say></Response>")
    assert "break" not in out
    assert "<Say voice=" in out and "</Say>" in out

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { scan, stripSayChildren, verdict } from './twilio-twiml-schema-audit.mjs';

const GOOD = `<?xml version="1.0" encoding="UTF-8"?>
<Response><Gather numDigits="4" action="/entered"><Say>Enter your code</Say></Gather></Response>`;

test('a correct document produces nothing', () => {
  assert.deepEqual(scan(GOOD), []);
  assert.equal(verdict(scan(GOOD), 0)[0], 'unexplained');
});

test('ssml inside say is not a casing error', () => {
  const doc = '<Response><Say>One<break time="500ms"/>two<say-as>3</say-as></Say></Response>';
  assert.deepEqual(scan(doc), []);
});

test('a lowercase say is still caught even though say is exempt', () => {
  const findings = scan('<Response><say>hello<break/></say></Response>');
  assert.ok(findings.some(([k, f, s]) => k === 'verb-casing' && f === 'say' && s === 'Say'));
  assert.ok(!findings.some(([, f]) => f === 'break'));
});

test('a miscased attribute names the camelCase form', () => {
  const [state, detail] = verdict(scan('<Response><Gather numdigits="4"/></Response>'), 12);
  assert.equal(state, 'attribute-casing');
  assert.match(detail, /numdigits should be numDigits/);
  assert.match(detail, /12 alert\(s\)/);
});

test('an unknown verb is not reported as a casing slip', () => {
  const [state, detail] = verdict(scan('<Response><Speak>hi</Speak></Response>'));
  assert.equal(state, 'unknown-verb');
  assert.match(detail, /not in the TwiML vocabulary/);
});

test('a root that is not Response is its own state', () => {
  const [state, detail] = verdict(scan('<Twiml><Say>hi</Say></Twiml>'));
  assert.equal(state, 'bad-root');
  assert.match(detail, /<Response>/);
});

test('a lowercase root is a casing finding, not a bad root', () => {
  const findings = scan('<response><Say>hi</Say></response>');
  assert.ok(findings.some(([k, f, s]) => k === 'verb-casing' && f === 'response' && s === 'Response'));
});

test('stripSayChildren keeps the tags it removes the contents of', () => {
  const out = stripSayChildren('<Response><Say voice="alice"><break/></Say></Response>');
  assert.ok(!out.includes('break'));
  assert.ok(out.includes('<Say voice=') && out.includes('</Say>'));
});