32011: Twilio cannot reach your SIP infrastructure

"""Report Twilio SIP trunks whose origination path explains a 32011.

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 place calls and
spend money.
"""
import argparse
import datetime as dt
import logging
import os
import sys

import requests

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

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

SIP_COMMS = 32011


def sip_host(sip_url):
    """Reduce a sip_url to its lowercase hostname.

    Three origination URIs that differ only in port or transport are three rows
    in the console and one machine on the network. Comparing hostnames is what
    tells those apart; comparing the raw strings never will.

    A value with no sip: or sips: scheme is not a SIP URI, so it reduces to ""
    and is reported rather than quietly treated as a hostname.
    """
    v = str(sip_url or "").strip()
    low = v.lower()
    for scheme in ("sips:", "sip:"):
        if low.startswith(scheme):
            v = v[len(scheme):]
            break
    else:
        return ""
    v = v.split(";", 1)[0].split("?", 1)[0]
    if "@" in v:
        v = v.rsplit("@", 1)[1]
    return v.split(":", 1)[0].strip().lower()


def transport_of(sip_url):
    """The transport a sip_url asks for: tls, tcp, udp, or "" when unstated.

    Transport lives inside the URI string as a parameter, or is implied by the
    sips: scheme. It is not a field on the resource, so nothing but this
    function will ever compare it against the trunk's secure flag.
    """
    v = str(sip_url or "").strip().lower()
    if v.startswith("sips:"):
        return "tls"
    for part in v.split(";")[1:]:
        name, _, value = part.partition("=")
        if name.strip() == "transport":
            return value.strip().split("?", 1)[0]
    return ""


def verdict(trunk, origination, alerts=0):
    """Classify one trunk's origination path. Pure, so it tests offline.

    origination is the trunk's OriginationUrl list. alerts is how many 32011
    alerts were seen in the window, which changes what a healthy-looking
    topology means: diverse paths plus alerts points at the edge rather than at
    the configuration.

    Returns (state, detail).
    """
    live = [u for u in (origination or []) if u.get("enabled")]
    if not live:
        return ("no-enabled-uri",
                "no enabled origination URI: Twilio has no address to send an "
                "INVITE to, so every inbound call on this trunk fails and %d "
                "alert(s) is an undercount of the damage." % alerts)

    hosts = [sip_host(u.get("sip_url")) for u in live]
    if "" in hosts:
        return ("unparseable-uri",
                "an enabled origination URI has no hostname this script can "
                "read, which usually means the sip_url is malformed and Twilio "
                "cannot resolve it either.")

    if trunk.get("secure") and not any(transport_of(u.get("sip_url")) == "tls"
                                       for u in live):
        return ("transport-mismatch",
                "secure is true on the trunk but no enabled URI asks for TLS: "
                "the trunk requires an encrypted path to an address that does "
                "not offer one, which fails every call rather than some of them.")

    distinct = sorted(set(hosts))
    if len(live) == 1:
        return ("single-path",
                "one enabled origination URI (%s): the %d alert(s) in this "
                "window had no second address to try, so a firewall rule or a "
                "reboot on that host is a full outage."
                % (live[0].get("sip_url") or "?", alerts))

    if len(distinct) == 1:
        return ("one-host",
                "%d enabled origination URIs all resolving to %s: three rows in "
                "the console, one machine on the network, and nothing to fail "
                "over to when it stops answering." % (len(live), distinct[0]))

    priorities = {u.get("priority") for u in live}
    if len(priorities) == 1:
        return ("flat-priority",
                "%d enabled URIs across %d hosts all share one priority, so "
                "Twilio spreads traffic over them by weight rather than trying "
                "them in order. That is load balancing, not failover."
                % (len(live), len(distinct)))

    if alerts:
        return ("reachability",
                "%d alert(s) against %d ordered URIs across %d hosts: the "
                "topology is not the problem, so look at the firewall ranges, "
                "the TLS version on the endpoint, and whether the PBX is "
                "answering with a 5xx." % (alerts, len(live), len(distinct)))

    return ("redundant",
            "%d enabled URIs across %d hosts with distinct priorities and no "
            "32011 in this window." % (len(live), len(distinct)))


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 page_meta(session, url, key, limit=100000, **params):
    """Page an API that carries an absolute meta.next_page_url."""
    params.setdefault("PageSize", 100)
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get(key, []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def sweep_alerts(session, since, limit, levels):
    """Both log levels, merged on sid.

    Several voice failures are logged at warning rather than error. Filtering to
    error alone reports a clean account while the trunk keeps failing.
    """
    seen = {}
    for level in levels:
        url = MONITOR + "/Alerts"
        params = {"LogLevel": level, "StartDate": since, "PageSize": 1000}
        got = 0
        while url and got < limit:
            page = get(session, url, **params)
            for a in page.get("alerts", []):
                seen.setdefault(a.get("sid"), a)
                got += 1
            url = (page.get("meta") or {}).get("next_page_url")
            params = {}
    return list(seen.values())


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=3,
                    help="how far back to sweep (alerts are retained 30 days)")
    ap.add_argument("--max-alerts", type=int, default=10000,
                    help="stop after this many alerts per log level")
    ap.add_argument("--errors-only", action="store_true",
                    help="skip the warning level, which will under-report")
    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)

    days = min(args.days, 30)
    since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    levels = ["error"] if args.errors_only else ["error", "warning"]

    alerts = sweep_alerts(session, since, args.max_alerts, levels)
    hits = [a for a in alerts
            if str(a.get("error_code") or "").strip() == str(SIP_COMMS)]

    trunks = page_meta(session, TRUNKING + "/Trunks", "trunks")
    if not trunks:
        log.info("no SIP trunks on this account")
        return 0

    bad = 0
    for t in trunks:
        origination = page_meta(
            session, "%s/Trunks/%s/OriginationUrls" % (TRUNKING, t.get("sid")),
            "origination_urls")
        state, detail = verdict(t, origination, len(hits))
        name = t.get("friendly_name") or t.get("domain_name") or t.get("sid")
        line = "%-18s %s  %s" % (state, name, detail)
        if state == "redundant":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        for u in origination:
            log.warning("    %-5s priority=%s weight=%s %s",
                        "on" if u.get("enabled") else "off", u.get("priority"),
                        u.get("weight"), u.get("sip_url"))
        log.warning("  repair: allowlist Twilio's SIP signalling and media "
                    "ranges, confirm the endpoint negotiates TLS 1.2, and add a "
                    "second origination URI on a different host with a higher "
                    "priority number")

    log.info("%d trunk(s), %d alert(s) with error_code %d in the last %d day(s)",
             len(trunks), len(hits), SIP_COMMS, days)
    return 1 if (bad or hits) else 0


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

/**
 * Report Twilio SIP trunks whose origination path explains a 32011.
 *
 * 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 TRUNKING = 'https://trunking.twilio.com/v1';
const MONITOR = 'https://monitor.twilio.com/v1';

const SIP_COMMS = 32011;

/**
 * Reduce a sip_url to its lowercase hostname. Three URIs that differ only in
 * port or transport are three rows in the console and one machine on the
 * network, and only a hostname comparison tells those apart. A value with no
 * sip: or sips: scheme is not a SIP URI, so it reduces to '' and is reported
 * rather than quietly treated as a hostname.
 */
export function sipHost(sipUrl) {
  let v = String(sipUrl ?? '').trim();
  const low = v.toLowerCase();
  let matched = false;
  for (const scheme of ['sips:', 'sip:']) {
    if (low.startsWith(scheme)) { v = v.slice(scheme.length); matched = true; break; }
  }
  if (!matched) return '';
  v = v.split(';')[0].split('?')[0];
  if (v.includes('@')) v = v.slice(v.lastIndexOf('@') + 1);
  return v.split(':')[0].trim().toLowerCase();
}

/**
 * The transport a sip_url asks for: tls, tcp, udp, or '' when unstated.
 * Transport is a URI parameter rather than a field on the resource, so nothing
 * but this compares it against the trunk's secure flag.
 */
export function transportOf(sipUrl) {
  const v = String(sipUrl ?? '').trim().toLowerCase();
  if (v.startsWith('sips:')) return 'tls';
  for (const part of v.split(';').slice(1)) {
    const eq = part.indexOf('=');
    if (eq === -1) continue;
    if (part.slice(0, eq).trim() === 'transport') {
      return part.slice(eq + 1).trim().split('?')[0];
    }
  }
  return '';
}

/**
 * Classify one trunk's origination path. `alerts` is how many 32011 alerts were
 * seen in the window, which changes what a healthy topology means. Pure.
 * Returns [state, detail].
 */
export function verdict(trunk, origination, alerts = 0) {
  const live = (origination ?? []).filter((u) => u.enabled);
  if (live.length === 0) {
    return ['no-enabled-uri',
      'no enabled origination URI: Twilio has no address to send an INVITE to, ' +
      `so every inbound call on this trunk fails and ${alerts} alert(s) is an ` +
      'undercount of the damage.'];
  }

  const hosts = live.map((u) => sipHost(u.sip_url));
  if (hosts.includes('')) {
    return ['unparseable-uri',
      'an enabled origination URI has no hostname this script can read, which ' +
      'usually means the sip_url is malformed and Twilio cannot resolve it either.'];
  }

  if (trunk.secure && !live.some((u) => transportOf(u.sip_url) === 'tls')) {
    return ['transport-mismatch',
      'secure is true on the trunk but no enabled URI asks for TLS: the trunk ' +
      'requires an encrypted path to an address that does not offer one, which ' +
      'fails every call rather than some of them.'];
  }

  const distinct = [...new Set(hosts)].sort();
  if (live.length === 1) {
    return ['single-path',
      `one enabled origination URI (${live[0].sip_url ?? '?'}): the ${alerts} ` +
      'alert(s) in this window had no second address to try, so a firewall rule ' +
      'or a reboot on that host is a full outage.'];
  }

  if (distinct.length === 1) {
    return ['one-host',
      `${live.length} enabled origination URIs all resolving to ${distinct[0]}: ` +
      'three rows in the console, one machine on the network, and nothing to ' +
      'fail over to when it stops answering.'];
  }

  if (new Set(live.map((u) => u.priority)).size === 1) {
    return ['flat-priority',
      `${live.length} enabled URIs across ${distinct.length} hosts all share one ` +
      'priority, so Twilio spreads traffic over them by weight rather than trying ' +
      'them in order. That is load balancing, not failover.'];
  }

  if (alerts) {
    return ['reachability',
      `${alerts} alert(s) against ${live.length} ordered URIs across ` +
      `${distinct.length} hosts: the topology is not the problem, so look at the ` +
      'firewall ranges, the TLS version on the endpoint, and whether the PBX is ' +
      'answering with a 5xx.'];
  }

  return ['redundant',
    `${live.length} enabled URIs across ${distinct.length} hosts with distinct ` +
    'priorities and no 32011 in this window.'];
}

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

/** Page an API that carries an absolute meta.next_page_url. */
export async function pageMeta(auth, url, key, params = {}) {
  let next = url;
  let query = { PageSize: 100, ...params };
  const out = [];
  while (next) {
    const page = await get(auth, next, query);
    out.push(...(page[key] ?? []));
    next = page.meta?.next_page_url ?? null;
    query = {};
  }
  return out;
}

/** Both log levels, merged on sid: several voice failures are warnings. */
export async function sweepAlerts(auth, since, limit, levels) {
  const seen = new Map();
  for (const level of levels) {
    let url = `${MONITOR}/Alerts`;
    let params = { LogLevel: level, StartDate: since, PageSize: 1000 };
    let got = 0;
    while (url && got < limit) {
      const page = await get(auth, url, params);
      for (const a of page.alerts ?? []) {
        if (!seen.has(a.sid)) seen.set(a.sid, a);
        got += 1;
      }
      url = page.meta?.next_page_url ?? null;
      params = {};
    }
  }
  return [...seen.values()];
}

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 days = Math.min(flagValue('--days', 3), 30);
  const levels = process.argv.includes('--errors-only') ? ['error'] : ['error', 'warning'];
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);

  const alerts = await sweepAlerts(auth, since, 10000, levels);
  const hits = alerts.filter(
    (a) => String(a.error_code ?? '').trim() === String(SIP_COMMS));

  const trunks = await pageMeta(auth, `${TRUNKING}/Trunks`, 'trunks');
  if (trunks.length === 0) {
    console.log('no SIP trunks on this account');
    return;
  }

  let bad = 0;
  for (const t of trunks) {
    const origination = await pageMeta(
      auth, `${TRUNKING}/Trunks/${t.sid}/OriginationUrls`, 'origination_urls');
    const [state, detail] = verdict(t, origination, hits.length);
    const name = t.friendly_name || t.domain_name || t.sid;
    const line = `${state.padEnd(18)} ${name}  ${detail}`;
    if (state === 'redundant') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    for (const u of origination) {
      console.warn(`    ${(u.enabled ? 'on' : 'off').padEnd(5)} ` +
                   `priority=${u.priority} weight=${u.weight} ${u.sip_url}`);
    }
    console.warn("  repair: allowlist Twilio's SIP signalling and media ranges, " +
                 'confirm the endpoint negotiates TLS 1.2, and add a second ' +
                 'origination URI on a different host with a higher priority number');
  }

  console.log(`${trunks.length} trunk(s), ${hits.length} alert(s) with error_code ` +
              `${SIP_COMMS} in the last ${days} day(s)`);
  process.exitCode = (bad || hits.length) ? 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_trunk_origination_audit import sip_host, transport_of, verdict


def test_sip_host_ignores_scheme_port_and_parameters():
    assert sip_host("sip:PBX.example.com:5060;transport=udp") == "pbx.example.com"
    assert sip_host("sips:pbx.example.com") == "pbx.example.com"
    assert sip_host("sip:trunk@pbx.example.com") == "pbx.example.com"
    # A bare host is not a SIP URI, so it reduces to nothing and gets reported.
    assert sip_host("pbx.example.com") == ""
    assert sip_host("") == ""


def test_transport_is_read_from_the_parameter_or_the_scheme():
    assert transport_of("sip:pbx.example.com;transport=TLS") == "tls"
    assert transport_of("sips:pbx.example.com") == "tls"
    assert transport_of("sip:pbx.example.com;transport=tcp") == "tcp"
    assert transport_of("sip:pbx.example.com") == ""


def test_no_enabled_uri_is_the_first_thing_reported():
    state, detail = verdict({}, [{"sip_url": "sip:a.example.com", "enabled": False}], 9)
    assert state == "no-enabled-uri"
    assert "9 alert(s)" in detail


def test_three_uris_on_one_host_is_not_redundancy():
    # The finding the console view argues against: three rows, one machine.
    origination = [
        {"sip_url": "sip:pbx.example.com:5060", "enabled": True, "priority": 10},
        {"sip_url": "sip:pbx.example.com:5061", "enabled": True, "priority": 20},
        {"sip_url": "sip:PBX.example.com;transport=tcp", "enabled": True, "priority": 30},
    ]
    state, detail = verdict({}, origination, 4)
    assert state == "one-host"
    assert "pbx.example.com" in detail


def test_secure_trunk_with_no_tls_uri_fails_every_call():
    origination = [{"sip_url": "sip:a.example.com;transport=udp", "enabled": True,
                    "priority": 10},
                   {"sip_url": "sip:b.example.com;transport=udp", "enabled": True,
                    "priority": 20}]
    state, detail = verdict({"secure": True}, origination, 0)
    assert state == "transport-mismatch"
    assert "every call" in detail


def test_a_secure_trunk_with_one_tls_uri_is_not_a_mismatch():
    origination = [{"sip_url": "sips:a.example.com", "enabled": True, "priority": 10},
                   {"sip_url": "sip:b.example.com", "enabled": True, "priority": 20}]
    assert verdict({"secure": True}, origination, 0)[0] == "redundant"


def test_one_enabled_uri_carries_the_alert_count():
    origination = [{"sip_url": "sip:a.example.com", "enabled": True, "priority": 10},
                   {"sip_url": "sip:b.example.com", "enabled": False, "priority": 20}]
    state, detail = verdict({}, origination, 12)
    assert state == "single-path"
    assert "12 alert(s)" in detail


def test_equal_priorities_are_load_balancing_not_failover():
    origination = [{"sip_url": "sip:a.example.com", "enabled": True, "priority": 10},
                   {"sip_url": "sip:b.example.com", "enabled": True, "priority": 10}]
    state, detail = verdict({}, origination, 0)
    assert state == "flat-priority"
    assert "not failover" in detail


def test_a_good_topology_with_alerts_points_at_the_edge():
    origination = [{"sip_url": "sip:a.example.com", "enabled": True, "priority": 10},
                   {"sip_url": "sip:b.example.com", "enabled": True, "priority": 20}]
    state, detail = verdict({}, origination, 31)
    assert state == "reachability"
    assert "TLS version" in detail


def test_a_malformed_uri_is_reported_rather_than_silently_dropped():
    origination = [{"sip_url": "pbx.example.com", "enabled": True, "priority": 10},
                   {"sip_url": "sip:b.example.com", "enabled": True, "priority": 20}]
    assert verdict({}, origination, 0)[0] == "unparseable-uri"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { sipHost, transportOf, verdict } from './twilio-trunk-origination-audit.mjs';

test('sipHost ignores scheme, port and parameters', () => {
  assert.equal(sipHost('sip:PBX.example.com:5060;transport=udp'), 'pbx.example.com');
  assert.equal(sipHost('sips:pbx.example.com'), 'pbx.example.com');
  assert.equal(sipHost('sip:trunk@pbx.example.com'), 'pbx.example.com');
  // A bare host is not a SIP URI, so it reduces to nothing and gets reported.
  assert.equal(sipHost('pbx.example.com'), '');
  assert.equal(sipHost(''), '');
});

test('transport is read from the parameter or the scheme', () => {
  assert.equal(transportOf('sip:pbx.example.com;transport=TLS'), 'tls');
  assert.equal(transportOf('sips:pbx.example.com'), 'tls');
  assert.equal(transportOf('sip:pbx.example.com;transport=tcp'), 'tcp');
  assert.equal(transportOf('sip:pbx.example.com'), '');
});

test('no enabled uri is the first thing reported', () => {
  const [state, detail] = verdict({}, [{ sip_url: 'sip:a.example.com', enabled: false }], 9);
  assert.equal(state, 'no-enabled-uri');
  assert.match(detail, /9 alert/);
});

test('three uris on one host is not redundancy', () => {
  const origination = [
    { sip_url: 'sip:pbx.example.com:5060', enabled: true, priority: 10 },
    { sip_url: 'sip:pbx.example.com:5061', enabled: true, priority: 20 },
    { sip_url: 'sip:PBX.example.com;transport=tcp', enabled: true, priority: 30 },
  ];
  const [state, detail] = verdict({}, origination, 4);
  assert.equal(state, 'one-host');
  assert.match(detail, /pbx.example.com/);
});

test('secure trunk with no tls uri fails every call', () => {
  const origination = [
    { sip_url: 'sip:a.example.com;transport=udp', enabled: true, priority: 10 },
    { sip_url: 'sip:b.example.com;transport=udp', enabled: true, priority: 20 },
  ];
  const [state, detail] = verdict({ secure: true }, origination, 0);
  assert.equal(state, 'transport-mismatch');
  assert.match(detail, /every call/);
});

test('a secure trunk with one tls uri is not a mismatch', () => {
  const origination = [{ sip_url: 'sips:a.example.com', enabled: true, priority: 10 },
                       { sip_url: 'sip:b.example.com', enabled: true, priority: 20 }];
  assert.equal(verdict({ secure: true }, origination, 0)[0], 'redundant');
});

test('one enabled uri carries the alert count', () => {
  const origination = [{ sip_url: 'sip:a.example.com', enabled: true, priority: 10 },
                       { sip_url: 'sip:b.example.com', enabled: false, priority: 20 }];
  const [state, detail] = verdict({}, origination, 12);
  assert.equal(state, 'single-path');
  assert.match(detail, /12 alert/);
});

test('equal priorities are load balancing not failover', () => {
  const origination = [{ sip_url: 'sip:a.example.com', enabled: true, priority: 10 },
                       { sip_url: 'sip:b.example.com', enabled: true, priority: 10 }];
  const [state, detail] = verdict({}, origination, 0);
  assert.equal(state, 'flat-priority');
  assert.match(detail, /not failover/);
});

test('a good topology with alerts points at the edge', () => {
  const origination = [{ sip_url: 'sip:a.example.com', enabled: true, priority: 10 },
                       { sip_url: 'sip:b.example.com', enabled: true, priority: 20 }];
  const [state, detail] = verdict({}, origination, 31);
  assert.equal(state, 'reachability');
  assert.match(detail, /TLS version/);
});

test('a malformed uri is reported rather than silently dropped', () => {
  const origination = [{ sip_url: 'pbx.example.com', enabled: true, priority: 10 },
                       { sip_url: 'sip:b.example.com', enabled: true, priority: 20 }];
  assert.equal(verdict({}, origination, 0)[0], 'unparseable-uri');
});