number webhooks on http, a private address or a dev tunnel

"""Report Twilio webhook URLs that are cleartext, unroutable, or a dev tunnel.

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

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

# Every field on a number that can hold a URL Twilio will fetch or notify.
NUMBER_URL_FIELDS = ("voice_url", "voice_fallback_url", "sms_url",
                     "sms_fallback_url", "status_callback")

# The same idea on a TwiML App, which wins outright when its SID is set on a
# number, so a number-only audit clears endpoints it never looked at.
APP_URL_FIELDS = ("voice_url", "voice_fallback_url", "sms_url",
                  "sms_fallback_url", "status_callback", "sms_status_callback")

# Substrings, not exact hosts. These services move between apex domains often
# enough that pinning the full name dates the check within a year.
TUNNEL_MARKERS = ("ngrok", "trycloudflare", "loca.lt", "serveo", "localtunnel")

# Urgency, worst first. Unreachable is failing now; cleartext is working and
# leaking; a tunnel is working and counting down. The order is the report.
SEVERITY = ("unreachable", "cleartext", "tunnel", "unreadable", "unset", "ok")


def is_private_host(host):
    """True for a host Twilio cannot route to from the public internet. Pure.

    The boundary worth getting right is 172.16.0.0/12: 172.31.x.x is private
    and 172.32.x.x is not, and a check written by eye usually places that edge
    one octet away from where it belongs.
    """
    h = str(host or "").strip().lower()
    if not h:
        return False
    if h in ("localhost", "localhost.localdomain") or h.endswith(".localhost"):
        return True
    if h in ("::1", "0:0:0:0:0:0:0:1"):
        return True
    parts = h.split(".")
    if len(parts) != 4 or not all(p.isdigit() and len(p) <= 3 for p in parts):
        return False
    octets = [int(p) for p in parts]
    if any(o > 255 for o in octets):
        return False
    a, b = octets[0], octets[1]
    return (a == 10 or a == 127 or a == 0
            or (a == 172 and 16 <= b <= 31)
            or (a == 192 and b == 168)
            or (a == 169 and b == 254))


def classify_url(url):
    """Classify one configured webhook URL. Pure. Returns (state, detail).

    Host before scheme, deliberately. http://localhost:3000/voice is both
    cleartext and unroutable, and only one of those is costing anything today:
    the exposure is theoretical on an endpoint Twilio can never connect to.
    """
    raw = str(url or "").strip()
    if not raw:
        return ("unset", "no URL configured on this field")

    parts = urlsplit(raw)
    scheme = (parts.scheme or "").lower()
    host = (parts.hostname or "").lower()
    if not host or scheme not in ("http", "https"):
        return ("unreadable",
                "not an absolute http or https URL, so Twilio has nothing to "
                "fetch: %r" % raw[:80])

    if is_private_host(host):
        return ("unreachable",
                "%s is a loopback or private address. Twilio dials from the "
                "public internet, so no firewall or allowlist change makes this "
                "reachable: every request raises 11205 or 11210." % host)

    if any(m in host for m in TUNNEL_MARKERS):
        return ("tunnel",
                "%s is a development tunnel. It answers correctly while the "
                "session that created it is alive and stops the moment that "
                "laptop sleeps, with no deploy to blame." % host)

    if scheme == "http":
        return ("cleartext",
                "http means the request body and the X-Twilio-Signature header "
                "cross the internet in clear. The signature proves origin, it "
                "does not encrypt: the caller number, the message body and the "
                "signature itself are all readable on the path.")

    return ("ok", "https on a public hostname")


def audit(resource, fields):
    """Classify every URL field on one number or app. Pure.

    Returns a list of (field, state, detail), with the healthy and unset fields
    kept in: the caller decides what to print, and dropping them here would make
    it impossible to say that an object was checked and was fine.
    """
    return [(f,) + classify_url(resource.get(f)) for f in fields]


def worst(findings):
    """The most urgent state among a resource's fields. Pure.

    A number can be cleartext on one field and unreachable on another, and the
    line it gets in the report should lead with the one that is failing now.
    """
    states = {state for _f, state, _d in findings}
    for state in SEVERITY:
        if state in states:
            return state
    return "ok"


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


def report(label, resource, fields, sid_field="sid"):
    """Print one object's findings. Returns 1 when anything needed flagging."""
    findings = audit(resource, fields)
    state = worst(findings)
    if state in ("ok", "unset"):
        log.info("%-12s %s  every URL field is https on a public hostname",
                 state, label)
        return 0
    log.warning("%-12s %s", state, label)
    for field, fstate, detail in findings:
        if fstate in ("ok", "unset"):
            continue
        log.warning("  %s: %s  %s", field, fstate, detail)
    log.warning("  repair: set the field to https://{public-host}/... on %s %s. "
                "When an Application SID is attached to a number, the app's "
                "URLs are the ones that win.",
                "app" if sid_field == "app" else "number",
                resource.get("sid") or "?")
    return 1


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-numbers", type=int, default=1000,
                    help="stop after this many phone numbers")
    ap.add_argument("--max-apps", type=int, default=1000,
                    help="stop after this many TwiML applications")
    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)

    numbers = page_all(session, "/Accounts/%s/IncomingPhoneNumbers.json" % account,
                       "incoming_phone_numbers", args.max_numbers)
    apps = page_all(session, "/Accounts/%s/Applications.json" % account,
                    "applications", args.max_apps)

    bad = 0
    for n in numbers:
        label = n.get("phone_number") or n.get("sid") or "?"
        bad += report(label, n, NUMBER_URL_FIELDS)
    for a in apps:
        label = "%s %s" % (a.get("sid") or "?", a.get("friendly_name") or "(unnamed)")
        bad += report(label, a, APP_URL_FIELDS, sid_field="app")

    log.info("%d number(s), %d app(s), %d with an insecure or unreachable "
             "webhook URL", len(numbers), len(apps), bad)
    return 1 if bad else 0


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

/**
 * Report Twilio webhook URLs that are cleartext, unroutable, or a dev tunnel.
 *
 * 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`;

// Every field on a number that can hold a URL Twilio will fetch or notify.
const NUMBER_URL_FIELDS = ['voice_url', 'voice_fallback_url', 'sms_url',
  'sms_fallback_url', 'status_callback'];

// The same on a TwiML App, whose URLs win outright when its SID is on a number.
const APP_URL_FIELDS = ['voice_url', 'voice_fallback_url', 'sms_url',
  'sms_fallback_url', 'status_callback', 'sms_status_callback'];

// Substrings, not exact hosts: these services change apex domains often enough
// that pinning the full name dates the check within a year.
const TUNNEL_MARKERS = ['ngrok', 'trycloudflare', 'loca.lt', 'serveo', 'localtunnel'];

// Urgency, worst first. Unreachable is failing now; cleartext is working and
// leaking; a tunnel is working and counting down.
const SEVERITY = ['unreachable', 'cleartext', 'tunnel', 'unreadable', 'unset', 'ok'];

/**
 * True for a host Twilio cannot route to from the public internet. Pure. The
 * boundary worth getting right is 172.16.0.0/12: 172.31 is private, 172.32 is
 * not, and a range written by eye usually misplaces that edge.
 */
export function isPrivateHost(host) {
  const h = String(host ?? '').trim().toLowerCase();
  if (!h) return false;
  if (h === 'localhost' || h === 'localhost.localdomain' || h.endsWith('.localhost')) {
    return true;
  }
  if (h === '::1' || h === '0:0:0:0:0:0:0:1') return true;
  const parts = h.split('.');
  if (parts.length !== 4 || !parts.every((p) => /^[0-9]{1,3}$/.test(p))) return false;
  const o = parts.map(Number);
  if (o.some((x) => x > 255)) return false;
  const [a, b] = o;
  return (a === 10 || a === 127 || a === 0
    || (a === 172 && b >= 16 && b <= 31)
    || (a === 192 && b === 168)
    || (a === 169 && b === 254));
}

/**
 * Classify one configured webhook URL. Pure. Returns [state, detail].
 * Host before scheme: http://localhost:3000/voice is both cleartext and
 * unroutable, and only the outage is costing anything today.
 */
export function classifyUrl(url) {
  const raw = String(url ?? '').trim();
  if (!raw) return ['unset', 'no URL configured on this field'];

  let u;
  try {
    u = new URL(raw);
  } catch {
    return ['unreadable',
      'not an absolute http or https URL, so Twilio has nothing to fetch: ' +
      raw.slice(0, 80)];
  }
  const scheme = u.protocol.replace(':', '').toLowerCase();
  const host = u.hostname.toLowerCase();
  if (!host || (scheme !== 'http' && scheme !== 'https')) {
    return ['unreadable',
      'not an absolute http or https URL, so Twilio has nothing to fetch: ' +
      raw.slice(0, 80)];
  }

  if (isPrivateHost(host)) {
    return ['unreachable',
      `${host} is a loopback or private address. Twilio dials from the public ` +
      'internet, so no firewall or allowlist change makes this reachable: ' +
      'every request raises 11205 or 11210.'];
  }

  if (TUNNEL_MARKERS.some((m) => host.includes(m))) {
    return ['tunnel',
      `${host} is a development tunnel. It answers correctly while the session ` +
      'that created it is alive and stops the moment that laptop sleeps, with ' +
      'no deploy to blame.'];
  }

  if (scheme === 'http') {
    return ['cleartext',
      'http means the request body and the X-Twilio-Signature header cross the ' +
      'internet in clear. The signature proves origin, it does not encrypt: ' +
      'the caller number, the message body and the signature itself are all ' +
      'readable on the path.'];
  }

  return ['ok', 'https on a public hostname'];
}

/**
 * Classify every URL field on one number or app. Pure. Healthy and unset fields
 * stay in, so a caller can still say an object was checked and was fine.
 */
export function audit(resource, fields) {
  return fields.map((f) => [f, ...classifyUrl(resource[f])]);
}

/** The most urgent state among a resource's fields. Pure. */
export function worst(findings) {
  const states = new Set(findings.map(([, state]) => state));
  for (const state of SEVERITY) if (states.has(state)) return state;
  return 'ok';
}

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 pageAll(auth, path, key, limit = 1000) {
  let url = BASE + path;
  let params = { PageSize: 1000 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page[key] ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

function report(label, resource, fields, kind) {
  const findings = audit(resource, fields);
  const state = worst(findings);
  if (state === 'ok' || state === 'unset') {
    console.log(`${state.padEnd(12)} ${label}  every URL field is https on a public hostname`);
    return 0;
  }
  console.warn(`${state.padEnd(12)} ${label}`);
  for (const [field, fstate, detail] of findings) {
    if (fstate === 'ok' || fstate === 'unset') continue;
    console.warn(`  ${field}: ${fstate}  ${detail}`);
  }
  console.warn(`  repair: set the field to https://{public-host}/... on ${kind} ` +
    `${resource.sid ?? '?'}. When an Application SID is attached to a number, ` +
    "the app's URLs are the ones that win.");
  return 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 numbers = await pageAll(auth, `/Accounts/${account}/IncomingPhoneNumbers.json`,
    'incoming_phone_numbers');
  const apps = await pageAll(auth, `/Accounts/${account}/Applications.json`,
    'applications');

  let bad = 0;
  for (const n of numbers) {
    bad += report(n.phone_number ?? n.sid ?? '?', n, NUMBER_URL_FIELDS, 'number');
  }
  for (const a of apps) {
    bad += report(`${a.sid ?? '?'} ${a.friendly_name ?? '(unnamed)'}`, a,
      APP_URL_FIELDS, 'app');
  }

  console.log(`${numbers.length} number(s), ${apps.length} app(s), ${bad} with ` +
    'an insecure or unreachable webhook URL');
  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_webhook_url_audit import (
    NUMBER_URL_FIELDS, audit, classify_url, is_private_host, worst)


def test_https_on_a_public_host_is_ok():
    state, detail = classify_url("https://hooks.example.com/voice")
    assert state == "ok"
    assert "public hostname" in detail


def test_http_is_reported_as_a_cleartext_signature():
    state, detail = classify_url("http://hooks.example.com/voice")
    assert state == "cleartext"
    assert "X-Twilio-Signature" in detail


def test_private_and_loopback_hosts_are_unreachable():
    for url in ("https://localhost:3000/voice", "https://127.0.0.1/voice",
                "https://10.0.4.31/sms", "https://192.168.1.20/sms",
                "https://172.16.0.9/sms", "https://169.254.169.254/voice"):
        assert classify_url(url)[0] == "unreachable", url


def test_the_172_boundary_is_where_the_rfc_puts_it():
    # 172.31 is private and 172.32 is not. A check that slides this edge either
    # clears a number that has never worked or condemns one that does.
    assert is_private_host("172.31.255.255") is True
    assert is_private_host("172.32.0.1") is False
    assert is_private_host("172.15.0.1") is False


def test_tunnel_hosts_are_their_own_finding():
    for url in ("https://ab12cd.ngrok.io/voice",
                "https://tall-cat-runs.trycloudflare.com/sms",
                "https://demo.loca.lt/voice"):
        state, detail = classify_url(url)
        assert state == "tunnel", url
        assert "laptop sleeps" in detail


def test_an_unreachable_host_over_http_leads_with_the_outage():
    # Both faults are present. Only one of them is costing anything today.
    assert classify_url("http://localhost:3000/voice")[0] == "unreachable"


def test_a_blank_field_is_unset_and_a_relative_path_is_unreadable():
    assert classify_url("")[0] == "unset"
    assert classify_url(None)[0] == "unset"
    assert classify_url("/voice")[0] == "unreadable"
    assert classify_url("ftp://hooks.example.com/voice")[0] == "unreadable"


def test_worst_ranks_the_outage_above_the_exposure():
    number = {"voice_url": "http://hooks.example.com/voice",
              "sms_url": "https://10.0.4.31/sms",
              "voice_fallback_url": "https://hooks.example.com/fallback"}
    findings = audit(number, NUMBER_URL_FIELDS)
    assert worst(findings) == "unreachable"
    assert ("voice_url", "cleartext") == findings[0][:2]


def test_a_fully_healthy_number_reports_ok():
    number = {"voice_url": "https://hooks.example.com/voice",
              "sms_url": "https://hooks.example.com/sms"}
    assert worst(audit(number, NUMBER_URL_FIELDS)) == "unset"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  audit, classifyUrl, isPrivateHost, worst,
} from './twilio-webhook-url-audit.mjs';

const NUMBER_URL_FIELDS = ['voice_url', 'voice_fallback_url', 'sms_url',
  'sms_fallback_url', 'status_callback'];

test('https on a public host is ok', () => {
  const [state, detail] = classifyUrl('https://hooks.example.com/voice');
  assert.equal(state, 'ok');
  assert.match(detail, /public hostname/);
});

test('http is reported as a cleartext signature', () => {
  const [state, detail] = classifyUrl('http://hooks.example.com/voice');
  assert.equal(state, 'cleartext');
  assert.match(detail, /X-Twilio-Signature/);
});

test('private and loopback hosts are unreachable', () => {
  for (const url of ['https://localhost:3000/voice', 'https://127.0.0.1/voice',
    'https://10.0.4.31/sms', 'https://192.168.1.20/sms',
    'https://172.16.0.9/sms', 'https://169.254.169.254/voice']) {
    assert.equal(classifyUrl(url)[0], 'unreachable', url);
  }
});

test('the 172 boundary is where the RFC puts it', () => {
  assert.equal(isPrivateHost('172.31.255.255'), true);
  assert.equal(isPrivateHost('172.32.0.1'), false);
  assert.equal(isPrivateHost('172.15.0.1'), false);
});

test('tunnel hosts are their own finding', () => {
  for (const url of ['https://ab12cd.ngrok.io/voice',
    'https://tall-cat-runs.trycloudflare.com/sms',
    'https://demo.loca.lt/voice']) {
    const [state, detail] = classifyUrl(url);
    assert.equal(state, 'tunnel', url);
    assert.match(detail, /laptop sleeps/);
  }
});

test('an unreachable host over http leads with the outage', () => {
  assert.equal(classifyUrl('http://localhost:3000/voice')[0], 'unreachable');
});

test('a blank field is unset and a relative path is unreadable', () => {
  assert.equal(classifyUrl('')[0], 'unset');
  assert.equal(classifyUrl(null)[0], 'unset');
  assert.equal(classifyUrl('/voice')[0], 'unreadable');
  assert.equal(classifyUrl('ftp://hooks.example.com/voice')[0], 'unreadable');
});

test('worst ranks the outage above the exposure', () => {
  const number = {
    voice_url: 'http://hooks.example.com/voice',
    sms_url: 'https://10.0.4.31/sms',
    voice_fallback_url: 'https://hooks.example.com/fallback',
  };
  const findings = audit(number, NUMBER_URL_FIELDS);
  assert.equal(worst(findings), 'unreachable');
  assert.deepEqual(findings[0].slice(0, 2), ['voice_url', 'cleartext']);
});

test('a fully healthy number reports unset for the fields it does not set', () => {
  const number = {
    voice_url: 'https://hooks.example.com/voice',
    sms_url: 'https://hooks.example.com/sms',
  };
  assert.equal(worst(audit(number, NUMBER_URL_FIELDS)), 'unset');
});