a SIP Domain with no auth_type accepts no traffic at all

"""Report Twilio SIP Domains that cannot accept traffic.

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 logging
import os
import sys

import requests

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

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

# The two authentication modes a SIP Domain can declare, and the key each one
# counts its mappings under in the dict handed to verdict().
COUNT_KEY = {"IP_ACL": "ip_acl", "CREDENTIAL_LIST": "credential_list"}


def auth_modes(domain):
    """Split auth_type into the modes it declares.

    A domain can carry both modes, comma separated, and the field arrives with
    inconsistent case and spacing. Comparing the raw string against one mode
    name reports a correctly configured both-modes domain as unrecognised.
    """
    raw = str(domain.get("auth_type") or "")
    return [m.strip().upper() for m in raw.replace(";", ",").split(",") if m.strip()]


def verdict(domain, mappings=None):
    """Classify one SIP Domain. Pure, so the rules can be tested without a
    network.

    mappings is {"credential_list": n, "ip_acl": n} for this domain, or None
    when the mapping subresources were not fetched. None means "not checked"
    and must not be read as "nothing mapped".

    Returns (state, detail).
    """
    modes = auth_modes(domain)
    if not modes:
        return ("inert",
                "auth_type is empty: a SIP Domain with no auth_type cannot "
                "receive any traffic. Every INVITE is refused at "
                "authentication, before voice_url is ever fetched.")

    unmapped = []
    if mappings is not None:
        unmapped = [m for m in modes if not mappings.get(COUNT_KEY.get(m, m), 0)]
        if len(unmapped) == len(modes):
            return ("auth-unmapped",
                    "auth_type declares %s but no credential list or IP ACL is "
                    "mapped to this domain, so there is nothing for a caller to "
                    "authenticate against." % "/".join(modes))

    if not str(domain.get("voice_url") or "").strip():
        return ("no-handler",
                "authentication is configured but voice_url is empty: the call "
                "is accepted and then has no instructions.")

    if unmapped:
        return ("partial-auth",
                "%s is declared with nothing mapped to it, so callers using "
                "that mode are refused while the other mode works. This is the "
                "one that reads as intermittent." % "/".join(unmapped))

    if not str(domain.get("voice_fallback_url") or "").strip():
        return ("no-fallback",
                "no voice_fallback_url: authenticated calls are dropped rather "
                "than rescued the moment your handler returns non-2xx.")

    return ("routed",
            "authenticated by %s, with a handler and a fallback"
            % ", ".join(modes))


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(session, url, key, params=None):
    """Page a 2010-04-01 list. next_page_uri here is a path, not a full URL."""
    params = dict(params or {})
    params.setdefault("PageSize", 100)
    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_domains(session, account):
    return page(session, "%s/Accounts/%s/SIP/Domains.json" % (BASE, account),
                "domains")


def mapping_counts(session, account, domain_sid):
    """How many credential lists and IP ACLs are mapped to this domain."""
    root = "%s/Accounts/%s/SIP/Domains/%s/Auth/Calls" % (BASE, account, domain_sid)
    creds = page(session, root + "/CredentialListMappings.json",
                 "credential_list_mappings")
    acls = page(session, root + "/IpAccessControlListMappings.json",
                "ip_access_control_list_mappings")
    return {"credential_list": len(creds), "ip_acl": len(acls)}


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--check-mappings", action="store_true",
                    help="two extra GETs per domain to confirm the declared "
                         "auth modes have something mapped to them")
    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)

    domains = list_domains(session, account)
    if not domains:
        log.info("no SIP domains on this account")
        return 0

    bad = 0
    for d in domains:
        mappings = None
        if args.check_mappings:
            mappings = mapping_counts(session, account, d.get("sid"))
        state, detail = verdict(d, mappings)
        name = d.get("domain_name") or d.get("friendly_name") or d.get("sid")
        line = "%-13s %s  %s" % (state, name, detail)
        if state == "routed":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if mappings is not None:
            log.warning("  mapped: %d credential list(s), %d IP ACL(s)",
                        mappings["credential_list"], mappings["ip_acl"])
        log.warning("  repair: POST %s/Accounts/%s/SIP/Domains/%s/Auth/Calls/"
                    "CredentialListMappings.json CredentialListSid=CLxxx "
                    "(or the IpAccessControlListMappings equivalent)",
                    BASE, account, d.get("sid"))

    log.info("%d SIP domain(s), %d unable to accept traffic", len(domains), bad)
    return 1 if bad else 0


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

/**
 * Report Twilio SIP Domains that cannot accept traffic.
 *
 * 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`;

// The two authentication modes a SIP Domain can declare, and the key each one
// counts its mappings under in the object handed to verdict().
const COUNT_KEY = { IP_ACL: 'ip_acl', CREDENTIAL_LIST: 'credential_list' };

/**
 * Split auth_type into the modes it declares. A domain can carry both, comma
 * separated, and the field arrives with inconsistent case and spacing.
 */
export function authModes(domain) {
  const raw = String(domain.auth_type ?? '');
  return raw.replace(/;/g, ',').split(',')
    .map((m) => m.trim().toUpperCase())
    .filter(Boolean);
}

/**
 * Classify one SIP Domain. Pure, so the rules can be tested without a network.
 * `mappings` is { credential_list, ip_acl } for this domain, or null when the
 * subresources were not fetched: null means "not checked", never "nothing
 * mapped". Returns [state, detail].
 */
export function verdict(domain, mappings = null) {
  const modes = authModes(domain);
  if (modes.length === 0) {
    return ['inert',
      'auth_type is empty: a SIP Domain with no auth_type cannot receive any ' +
      'traffic. Every INVITE is refused at authentication, before voice_url ' +
      'is ever fetched.'];
  }

  let unmapped = [];
  if (mappings !== null) {
    unmapped = modes.filter((m) => !(mappings[COUNT_KEY[m] ?? m] ?? 0));
    if (unmapped.length === modes.length) {
      return ['auth-unmapped',
        `auth_type declares ${modes.join('/')} but no credential list or IP ` +
        'ACL is mapped to this domain, so there is nothing for a caller to ' +
        'authenticate against.'];
    }
  }

  if (!String(domain.voice_url ?? '').trim()) {
    return ['no-handler',
      'authentication is configured but voice_url is empty: the call is ' +
      'accepted and then has no instructions.'];
  }

  if (unmapped.length) {
    return ['partial-auth',
      `${unmapped.join('/')} is declared with nothing mapped to it, so callers ` +
      'using that mode are refused while the other mode works. This is the one ' +
      'that reads as intermittent.'];
  }

  if (!String(domain.voice_fallback_url ?? '').trim()) {
    return ['no-fallback',
      'no voice_fallback_url: authenticated calls are dropped rather than ' +
      'rescued the moment your handler returns non-2xx.'];
  }

  return ['routed', `authenticated by ${modes.join(', ')}, with a handler and a fallback`];
}

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 list. next_page_uri here is a path, not a full URL. */
async function pageAll(auth, url, key) {
  let params = { PageSize: 100 };
  const out = [];
  while (url) {
    const body = await get(auth, url, params);
    out.push(...(body[key] ?? []));
    url = body.next_page_uri ? HOST + body.next_page_uri : null;
    params = {};
  }
  return out;
}

export async function listDomains(auth, account) {
  return pageAll(auth, `${BASE}/Accounts/${account}/SIP/Domains.json`, 'domains');
}

export async function mappingCounts(auth, account, domainSid) {
  const root = `${BASE}/Accounts/${account}/SIP/Domains/${domainSid}/Auth/Calls`;
  const creds = await pageAll(auth, `${root}/CredentialListMappings.json`,
                              'credential_list_mappings');
  const acls = await pageAll(auth, `${root}/IpAccessControlListMappings.json`,
                             'ip_access_control_list_mappings');
  return { credential_list: creds.length, ip_acl: acls.length };
}

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 checkMappings = process.argv.includes('--check-mappings');

  const domains = await listDomains(auth, account);
  if (domains.length === 0) {
    console.log('no SIP domains on this account');
    return;
  }

  let bad = 0;
  for (const d of domains) {
    const mappings = checkMappings ? await mappingCounts(auth, account, d.sid) : null;
    const [state, detail] = verdict(d, mappings);
    const name = d.domain_name || d.friendly_name || d.sid;
    const line = `${state.padEnd(13)} ${name}  ${detail}`;
    if (state === 'routed') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    if (mappings !== null) {
      console.warn(`  mapped: ${mappings.credential_list} credential list(s), ` +
                   `${mappings.ip_acl} IP ACL(s)`);
    }
    console.warn(`  repair: POST ${BASE}/Accounts/${account}/SIP/Domains/${d.sid}` +
                 '/Auth/Calls/CredentialListMappings.json CredentialListSid=CLxxx ' +
                 '(or the IpAccessControlListMappings equivalent)');
  }

  console.log(`${domains.length} SIP domain(s), ${bad} unable to accept traffic`);
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly, so importing this module from the test file
// does not execute main(), fail on missing credentials and set a non-zero exit
// code that fails the suite 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_domain_auth_audit import auth_modes, verdict

ROUTED = {"auth_type": "CREDENTIAL_LIST",
          "voice_url": "https://app.example.com/voice",
          "voice_fallback_url": "https://app.example.com/fallback"}


def test_empty_auth_type_is_inert():
    state, detail = verdict({"auth_type": "",
                             "voice_url": "https://app.example.com/voice"})
    assert state == "inert"
    assert "cannot receive any traffic" in detail


def test_missing_auth_type_reads_the_same_as_an_empty_one():
    assert verdict({"voice_url": "https://app.example.com/voice"})[0] == "inert"
    assert verdict({"auth_type": None})[0] == "inert"


def test_both_modes_comma_separated_are_parsed_as_two():
    # The reason auth_type is split rather than compared as a string.
    assert auth_modes({"auth_type": "ip_acl, CREDENTIAL_LIST"}) == \
        ["IP_ACL", "CREDENTIAL_LIST"]


def test_declared_but_nothing_mapped_is_auth_unmapped():
    state, _ = verdict(ROUTED, {"credential_list": 0, "ip_acl": 0})
    assert state == "auth-unmapped"


def test_not_checking_mappings_is_not_the_same_as_nothing_mapped():
    assert verdict(ROUTED)[0] == "routed"


def test_one_of_two_modes_unmapped_is_the_intermittent_case():
    domain = dict(ROUTED, auth_type="IP_ACL,CREDENTIAL_LIST")
    state, detail = verdict(domain, {"credential_list": 1, "ip_acl": 0})
    assert state == "partial-auth"
    assert "IP_ACL" in detail


def test_authenticated_domain_with_no_voice_url_is_no_handler():
    domain = dict(ROUTED, voice_url="")
    assert verdict(domain, {"credential_list": 1, "ip_acl": 0})[0] == "no-handler"


def test_missing_fallback_is_reported_after_the_bigger_failures():
    domain = dict(ROUTED, voice_fallback_url="")
    state, detail = verdict(domain, {"credential_list": 1, "ip_acl": 0})
    assert state == "no-fallback"
    assert "non-2xx" in detail

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

const ROUTED = {
  auth_type: 'CREDENTIAL_LIST',
  voice_url: 'https://app.example.com/voice',
  voice_fallback_url: 'https://app.example.com/fallback',
};

test('empty auth_type is inert', () => {
  const [state, detail] = verdict({ auth_type: '', voice_url: 'https://app.example.com/voice' });
  assert.equal(state, 'inert');
  assert.match(detail, /cannot receive any traffic/);
});

test('missing auth_type reads the same as an empty one', () => {
  assert.equal(verdict({ voice_url: 'https://app.example.com/voice' })[0], 'inert');
  assert.equal(verdict({ auth_type: null })[0], 'inert');
});

test('both modes comma separated are parsed as two', () => {
  assert.deepEqual(authModes({ auth_type: 'ip_acl, CREDENTIAL_LIST' }),
                   ['IP_ACL', 'CREDENTIAL_LIST']);
});

test('declared but nothing mapped is auth-unmapped', () => {
  assert.equal(verdict(ROUTED, { credential_list: 0, ip_acl: 0 })[0], 'auth-unmapped');
});

test('not checking mappings is not the same as nothing mapped', () => {
  assert.equal(verdict(ROUTED)[0], 'routed');
});

test('one of two modes unmapped is the intermittent case', () => {
  const domain = { ...ROUTED, auth_type: 'IP_ACL,CREDENTIAL_LIST' };
  const [state, detail] = verdict(domain, { credential_list: 1, ip_acl: 0 });
  assert.equal(state, 'partial-auth');
  assert.match(detail, /IP_ACL/);
});

test('authenticated domain with no voice_url is no-handler', () => {
  const domain = { ...ROUTED, voice_url: '' };
  assert.equal(verdict(domain, { credential_list: 1, ip_acl: 0 })[0], 'no-handler');
});

test('missing fallback is reported after the bigger failures', () => {
  const domain = { ...ROUTED, voice_fallback_url: '' };
  const [state, detail] = verdict(domain, { credential_list: 1, ip_acl: 0 });
  assert.equal(state, 'no-fallback');
  assert.match(detail, /non-2xx/);
});