Dial fails with 32009 because the SIP endpoint is not there

"""Report Twilio 32009 alerts and say why each SIP endpoint was unreachable.

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

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

NOT_REGISTERED = 32009


def sip_target(uri):
    """Split a SIP URI into (user, domain).

    The domain is lowercased because SIP hostnames are case insensitive. The
    user is not, and that is the entire point of this function: a credential
    created as Reception and a Dial aimed at reception are different endpoints,
    and a parser that folds case throws away the only evidence that says so.

    Handles sip: and sips:, a display name in angle brackets, a port, and URI
    parameters. Returns ("", "") when there is nothing to split.
    """
    v = str(uri or "").strip()
    if "<" in v and ">" in v:
        v = v[v.index("<") + 1:v.index(">")].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 "@" not in v:
        return ("", "")
    user, host = v.rsplit("@", 1)
    return (user.strip(), host.split(":", 1)[0].strip().lower())


def verdict(target, domains):
    """Explain one 32009. Pure, so the rules can be tested without a network.

    target is (user, domain) from sip_target. domains maps a lowercase
    domain_name to {"sip_registration": bool, "usernames": [...]}, assembled
    from the SIP Domains list and each domain's registration credential lists.

    Returns (state, detail).
    """
    user, host = target
    if not host:
        return ("unresolved",
                "no sip: destination on the failing leg, so the username cannot "
                "be compared against anything. Check the child call by hand.")

    domain = domains.get(host)
    if domain is None:
        return ("unknown-domain",
                "%s is not a SIP Domain on this account, so no endpoint can "
                "hold a registration on it and every Dial to it fails the same "
                "way." % host)

    if not domain.get("sip_registration"):
        return ("registration-off",
                "sip_registration is false on %s: the domain can accept INVITEs "
                "from mapped credentials but nothing may register to it, so "
                "sip:%s@%s has no registration to route to and never will."
                % (host, user, host))

    usernames = list(domain.get("usernames") or [])
    if not usernames:
        return ("no-credentials",
                "%s allows registration but no credential list is mapped to its "
                "Auth/Registrations subresource, so there is no username any "
                "endpoint could register with." % host)

    if user in usernames:
        return ("offline",
                "%s is a registerable credential on %s, so the username is "
                "right and the endpoint simply held no registration when the "
                "call arrived: a dropped REGISTER refresh, a closed softphone, "
                "or a NAT binding that expired." % (user, host))

    folded = {u.casefold(): u for u in usernames}
    if user.casefold() in folded:
        return ("case-mismatch",
                "the credential on %s is %s and the Dial asked for %s. SIP "
                "usernames are compared exactly, so these are two different "
                "endpoints however alike they read."
                % (host, folded[user.casefold()], user))

    return ("unknown-user",
            "%s is not among the %d registerable username(s) on %s, so this "
            "call was never going to connect regardless of who was online."
            % (user, len(usernames), host))


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_2010(session, url, key, **params):
    """Page a 2010-04-01 listing. next_page_uri here is a path, not a URL."""
    params.setdefault("PageSize", 1000)
    out = []
    while url:
        body = get(session, url, **params)
        out.extend(body.get(key, []))
        nxt = body.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out


def list_alerts(session, since, limit, log_level):
    """Page the Monitor alerts at one log level. next_page_url is absolute."""
    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 sweep_alerts(session, since, limit, levels):
    """Both log levels, merged on sid.

    Several voice failures are logged at warning rather than error. A sweep that
    reads only the error level reports a clean account while the calls keep
    failing, which is why this function takes a list of levels at all.
    """
    seen = {}
    for level in levels:
        for a in list_alerts(session, since, limit, level):
            seen.setdefault(a.get("sid"), a)
    return list(seen.values())


def registerable_domains(session, account):
    """Map each SIP domain to its registration flag and registerable usernames.

    The Auth/Registrations mapping is a different subresource from Auth/Calls. A
    credential list mapped only to the latter is correct, present, and unable to
    register anything, so reading the wrong one produces a confident wrong answer.
    """
    out = {}
    domains = page_2010(session, "%s/Accounts/%s/SIP/Domains.json" % (BASE, account),
                        "sip_domains")
    for d in domains:
        name = str(d.get("domain_name") or "").strip().lower()
        if not name:
            continue
        usernames = []
        if d.get("sip_registration"):
            mappings = page_2010(
                session,
                "%s/Accounts/%s/SIP/Domains/%s/Auth/Registrations/"
                "CredentialListMappings.json" % (BASE, account, d.get("sid")),
                "credential_list_mappings")
            for m in mappings:
                creds = page_2010(
                    session,
                    "%s/Accounts/%s/SIP/CredentialLists/%s/Credentials.json"
                    % (BASE, account, m.get("sid")), "credentials")
                usernames.extend(str(c.get("username") or "").strip() for c in creds)
        out[name] = {"sip_registration": bool(d.get("sip_registration")),
                     "usernames": [u for u in usernames if u]}
    return out


def sip_leg(session, account, parent_sid):
    """The child leg of a call whose destination is a SIP URI, or an empty string."""
    children = page_2010(session, "%s/Accounts/%s/Calls.json" % (BASE, account),
                         "calls", ParentCallSid=parent_sid)
    for c in children:
        to = str(c.get("to") or "").strip()
        if to.lower().startswith("sip:") or to.lower().startswith("sips:"):
            return to
    return ""


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=7,
                    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(NOT_REGISTERED)]
    if not hits:
        log.info("0 alert(s) with error_code %d in the last %d day(s)",
                 NOT_REGISTERED, days)
        return 0

    domains = registerable_domains(session, account)
    targets = {}
    counts = {}
    for a in hits:
        parent = str(a.get("resource_sid") or "")
        if not parent.startswith("CA"):
            log.warning("32009 alert %s has no call sid to resolve", a.get("sid"))
            continue
        if parent not in targets:
            targets[parent] = sip_target(sip_leg(session, account, parent))
        state, detail = verdict(targets[parent], domains)
        counts[state] = counts.get(state, 0) + 1
        log.warning("%-16s %s  %s", state, parent, detail)

    log.warning("%d alert(s) with error_code %d across %d call(s): %s",
                len(hits), NOT_REGISTERED, len(targets),
                ", ".join("%s=%d" % kv for kv in sorted(counts.items())))
    log.warning("  repair: make the username in <Sip> match a credential "
                "exactly, or set SipRegistration=true on the domain, or map the "
                "credential list to Auth/Registrations")
    log.warning("  live registrations: Console > Voice > Manage > SIP Domains > "
                "Registered SIP Endpoints")
    return 1


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

/**
 * Report Twilio 32009 alerts and say why each SIP endpoint was unreachable.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The repair is printed, never
 * performed.
 */
const HOST = 'https://api.twilio.com';
const BASE = `${HOST}/2010-04-01`;
const MONITOR = 'https://monitor.twilio.com/v1';

const NOT_REGISTERED = 32009;

/**
 * Split a SIP URI into [user, domain]. The domain is lowercased because SIP
 * hostnames are case insensitive; the user is not, because a credential created
 * as Reception and a Dial aimed at reception are different endpoints and folding
 * case throws away the evidence that says so.
 */
export function sipTarget(uri) {
  let v = String(uri ?? '').trim();
  if (v.includes('<') && v.includes('>')) {
    v = v.slice(v.indexOf('<') + 1, v.indexOf('>')).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('@')) return ['', ''];
  const at = v.lastIndexOf('@');
  const user = v.slice(0, at).trim();
  const host = v.slice(at + 1).split(':')[0].trim().toLowerCase();
  return [user, host];
}

/**
 * Explain one 32009. `target` is [user, domain] from sipTarget; `domains` maps a
 * lowercase domain_name to { sip_registration, usernames }. Pure. Returns
 * [state, detail].
 */
export function verdict(target, domains = {}) {
  const [user, host] = target;
  if (!host) {
    return ['unresolved',
      'no sip: destination on the failing leg, so the username cannot be ' +
      'compared against anything. Check the child call by hand.'];
  }

  const domain = domains[host];
  if (domain === undefined) {
    return ['unknown-domain',
      `${host} is not a SIP Domain on this account, so no endpoint can hold a ` +
      'registration on it and every Dial to it fails the same way.'];
  }

  if (!domain.sip_registration) {
    return ['registration-off',
      `sip_registration is false on ${host}: the domain can accept INVITEs from ` +
      `mapped credentials but nothing may register to it, so sip:${user}@${host} ` +
      'has no registration to route to and never will.'];
  }

  const usernames = domain.usernames ?? [];
  if (usernames.length === 0) {
    return ['no-credentials',
      `${host} allows registration but no credential list is mapped to its ` +
      'Auth/Registrations subresource, so there is no username any endpoint ' +
      'could register with.'];
  }

  if (usernames.includes(user)) {
    return ['offline',
      `${user} is a registerable credential on ${host}, so the username is right ` +
      'and the endpoint simply held no registration when the call arrived: a ' +
      'dropped REGISTER refresh, a closed softphone, or a NAT binding that expired.'];
  }

  const folded = new Map(usernames.map((u) => [u.toLowerCase(), u]));
  if (folded.has(user.toLowerCase())) {
    return ['case-mismatch',
      `the credential on ${host} is ${folded.get(user.toLowerCase())} and the ` +
      `Dial asked for ${user}. SIP usernames are compared exactly, so these are ` +
      'two different endpoints however alike they read.'];
  }

  return ['unknown-user',
    `${user} is not among the ${usernames.length} registerable username(s) on ` +
    `${host}, so this call was never going to connect regardless of who was online.`];
}

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 a 2010-04-01 listing. next_page_uri here is a path, not a URL. */
export async function page2010(auth, url, key, params = {}) {
  let next = url;
  let query = { PageSize: 1000, ...params };
  const out = [];
  while (next) {
    const body = await get(auth, next, query);
    out.push(...(body[key] ?? []));
    next = body.next_page_uri ? HOST + body.next_page_uri : null;
    query = {};
  }
  return out;
}

export async function listAlerts(auth, since, limit, logLevel) {
  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);
}

/** 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) {
    for (const a of await listAlerts(auth, since, limit, level)) {
      if (!seen.has(a.sid)) seen.set(a.sid, a);
    }
  }
  return [...seen.values()];
}

async function registerableDomains(auth, account) {
  const out = {};
  const domains = await page2010(
    auth, `${BASE}/Accounts/${account}/SIP/Domains.json`, 'sip_domains');
  for (const d of domains) {
    const name = String(d.domain_name ?? '').trim().toLowerCase();
    if (!name) continue;
    const usernames = [];
    if (d.sip_registration) {
      const mappings = await page2010(
        auth,
        `${BASE}/Accounts/${account}/SIP/Domains/${d.sid}/Auth/Registrations/` +
        'CredentialListMappings.json', 'credential_list_mappings');
      for (const m of mappings) {
        const creds = await page2010(
          auth,
          `${BASE}/Accounts/${account}/SIP/CredentialLists/${m.sid}/Credentials.json`,
          'credentials');
        for (const c of creds) {
          const u = String(c.username ?? '').trim();
          if (u) usernames.push(u);
        }
      }
    }
    out[name] = { sip_registration: Boolean(d.sip_registration), usernames };
  }
  return out;
}

async function sipLeg(auth, account, parentSid) {
  const children = await page2010(
    auth, `${BASE}/Accounts/${account}/Calls.json`, 'calls',
    { ParentCallSid: parentSid });
  for (const c of children) {
    const to = String(c.to ?? '').trim().toLowerCase();
    if (to.startsWith('sip:') || to.startsWith('sips:')) return String(c.to).trim();
  }
  return '';
}

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', 7), 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(NOT_REGISTERED));
  if (hits.length === 0) {
    console.log(`0 alert(s) with error_code ${NOT_REGISTERED} in the last ${days} day(s)`);
    return;
  }

  const domains = await registerableDomains(auth, account);
  const targets = new Map();
  const counts = new Map();
  for (const a of hits) {
    const parent = String(a.resource_sid ?? '');
    if (!parent.startsWith('CA')) {
      console.warn(`32009 alert ${a.sid} has no call sid to resolve`);
      continue;
    }
    if (!targets.has(parent)) {
      targets.set(parent, sipTarget(await sipLeg(auth, account, parent)));
    }
    const [state, detail] = verdict(targets.get(parent), domains);
    counts.set(state, (counts.get(state) ?? 0) + 1);
    console.warn(`${state.padEnd(16)} ${parent}  ${detail}`);
  }

  const summary = [...counts.entries()].sort().map(([k, v]) => `${k}=${v}`).join(', ');
  console.warn(`${hits.length} alert(s) with error_code ${NOT_REGISTERED} across ` +
               `${targets.size} call(s): ${summary}`);
  console.warn('  repair: make the username in <Sip> match a credential exactly, ' +
               'or set SipRegistration=true on the domain, or map the credential ' +
               'list to Auth/Registrations');
  console.warn('  live registrations: Console > Voice > Manage > SIP Domains > ' +
               'Registered SIP Endpoints');
  process.exitCode = 1;
}

// 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_sip_registration_audit import sip_target, verdict

DOMAINS = {
    "acme.sip.twilio.com": {"sip_registration": True,
                            "usernames": ["Reception", "warehouse"]},
    "calls-only.sip.twilio.com": {"sip_registration": False, "usernames": []},
    "open.sip.twilio.com": {"sip_registration": True, "usernames": []},
}


def test_plain_uri_splits_into_user_and_domain():
    assert sip_target("sip:warehouse@acme.sip.twilio.com") == \
        ("warehouse", "acme.sip.twilio.com")


def test_domain_is_lowercased_and_the_user_is_not():
    # Folding the user would destroy the only evidence the case-mismatch state
    # has to work with, so the asymmetry is deliberate and pinned here.
    assert sip_target("SIP:Reception@ACME.sip.twilio.com") == \
        ("Reception", "acme.sip.twilio.com")


def test_port_parameters_display_name_and_sips_all_reduce_the_same():
    assert sip_target("sips:warehouse@acme.sip.twilio.com:5061") == \
        ("warehouse", "acme.sip.twilio.com")
    assert sip_target("sip:warehouse@acme.sip.twilio.com;transport=tls") == \
        ("warehouse", "acme.sip.twilio.com")
    assert sip_target('"Front desk" <sip:warehouse@acme.sip.twilio.com>') == \
        ("warehouse", "acme.sip.twilio.com")


def test_a_tel_uri_or_a_bare_number_is_not_a_sip_target():
    assert sip_target("+15005550006") == ("", "")
    assert sip_target("sip:acme.sip.twilio.com") == ("", "")
    assert sip_target(None) == ("", "")


def test_missing_destination_is_unresolved_rather_than_a_guess():
    state, _ = verdict(("", ""), DOMAINS)
    assert state == "unresolved"


def test_domain_not_on_the_account_is_its_own_state():
    state, _ = verdict(("warehouse", "other.sip.twilio.com"), DOMAINS)
    assert state == "unknown-domain"


def test_registration_disabled_is_permanent_not_transient():
    state, detail = verdict(("warehouse", "calls-only.sip.twilio.com"), DOMAINS)
    assert state == "registration-off"
    assert "never will" in detail


def test_registration_enabled_with_nothing_mapped():
    state, detail = verdict(("warehouse", "open.sip.twilio.com"), DOMAINS)
    assert state == "no-credentials"
    assert "Auth/Registrations" in detail


def test_exact_match_means_the_endpoint_was_merely_offline():
    state, detail = verdict(("warehouse", "acme.sip.twilio.com"), DOMAINS)
    assert state == "offline"
    assert "REGISTER refresh" in detail


def test_case_mismatch_is_reported_separately_and_names_both_strings():
    # Reported as unknown-user, this sends someone to create a credential that
    # already exists. It is the whole reason the parser preserves case.
    state, detail = verdict(("reception", "acme.sip.twilio.com"), DOMAINS)
    assert state == "case-mismatch"
    assert "Reception" in detail
    assert "reception" in detail


def test_username_nobody_ever_created_is_unknown_user():
    state, detail = verdict(("nightshift", "acme.sip.twilio.com"), DOMAINS)
    assert state == "unknown-user"
    assert "2 registerable" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { sipTarget, verdict } from './twilio-sip-registration-audit.mjs';

const DOMAINS = {
  'acme.sip.twilio.com': { sip_registration: true, usernames: ['Reception', 'warehouse'] },
  'calls-only.sip.twilio.com': { sip_registration: false, usernames: [] },
  'open.sip.twilio.com': { sip_registration: true, usernames: [] },
};

test('plain uri splits into user and domain', () => {
  assert.deepEqual(sipTarget('sip:warehouse@acme.sip.twilio.com'),
                   ['warehouse', 'acme.sip.twilio.com']);
});

test('domain is lowercased and the user is not', () => {
  assert.deepEqual(sipTarget('SIP:Reception@ACME.sip.twilio.com'),
                   ['Reception', 'acme.sip.twilio.com']);
});

test('port, parameters, display name and sips all reduce the same', () => {
  assert.deepEqual(sipTarget('sips:warehouse@acme.sip.twilio.com:5061'),
                   ['warehouse', 'acme.sip.twilio.com']);
  assert.deepEqual(sipTarget('sip:warehouse@acme.sip.twilio.com;transport=tls'),
                   ['warehouse', 'acme.sip.twilio.com']);
  assert.deepEqual(sipTarget('"Front desk" <sip:warehouse@acme.sip.twilio.com>'),
                   ['warehouse', 'acme.sip.twilio.com']);
});

test('a tel uri or a bare number is not a sip target', () => {
  assert.deepEqual(sipTarget('+15005550006'), ['', '']);
  assert.deepEqual(sipTarget('sip:acme.sip.twilio.com'), ['', '']);
  assert.deepEqual(sipTarget(null), ['', '']);
});

test('missing destination is unresolved rather than a guess', () => {
  assert.equal(verdict(['', ''], DOMAINS)[0], 'unresolved');
});

test('domain not on the account is its own state', () => {
  assert.equal(verdict(['warehouse', 'other.sip.twilio.com'], DOMAINS)[0],
               'unknown-domain');
});

test('registration disabled is permanent not transient', () => {
  const [state, detail] = verdict(['warehouse', 'calls-only.sip.twilio.com'], DOMAINS);
  assert.equal(state, 'registration-off');
  assert.match(detail, /never will/);
});

test('registration enabled with nothing mapped', () => {
  const [state, detail] = verdict(['warehouse', 'open.sip.twilio.com'], DOMAINS);
  assert.equal(state, 'no-credentials');
  assert.match(detail, /Auth.Registrations/);
});

test('exact match means the endpoint was merely offline', () => {
  const [state, detail] = verdict(['warehouse', 'acme.sip.twilio.com'], DOMAINS);
  assert.equal(state, 'offline');
  assert.match(detail, /REGISTER refresh/);
});

test('case mismatch is reported separately and names both strings', () => {
  const [state, detail] = verdict(['reception', 'acme.sip.twilio.com'], DOMAINS);
  assert.equal(state, 'case-mismatch');
  assert.match(detail, /Reception/);
  assert.match(detail, /reception/);
});

test('username nobody ever created is unknown user', () => {
  const [state, detail] = verdict(['nightshift', 'acme.sip.twilio.com'], DOMAINS);
  assert.equal(state, 'unknown-user');
  assert.match(detail, /2 registerable/);
});