a published Studio Flow that no phone number points at

"""Report published Twilio Studio Flows that no phone number points at.

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

import requests

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

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


def attachments(flow_sid, numbers):
    """Find the numbers whose voice or SMS webhook runs this Flow. Pure.

    The attachment is a URL, not a reference: a number runs a Flow because its
    voice_url or sms_url is the Studio webhook for that FlowSid. Matched as a
    substring, because the URL can carry a query string and reconstructing the
    exact expected string is how a scan reports every Flow as unwired.

    Numbers whose voice_application_sid is set are collected separately: their
    voice_url is ignored at runtime, so this scan cannot answer for them.
    """
    out = {"voice": [], "sms": [], "via_application": []}
    if not flow_sid:
        return out
    for n in numbers or []:
        label = n.get("phone_number") or n.get("sid") or "?"
        if flow_sid in str(n.get("voice_url") or ""):
            out["voice"].append(label)
        if flow_sid in str(n.get("sms_url") or ""):
            out["sms"].append(label)
        if str(n.get("voice_application_sid") or "").strip():
            out["via_application"].append(label)
    return out


def verdict(flow, attach=None, executions=0):
    """Classify one published Flow's entry point. Pure, so the difference
    between "nothing can reach it" and "something reaches it from elsewhere"
    is written down rather than inferred.

    Returns (state, detail).
    """
    attach = attach or {"voice": [], "sms": [], "via_application": []}
    status = str(flow.get("status") or "").lower()
    wired = list(attach.get("voice") or []) + list(attach.get("sms") or [])
    executions = int(executions or 0)

    if status != "published":
        return ("unpublished",
                "status is %s, so there is no published definition for a number "
                "to run. Publish first; wiring a draft changes nothing."
                % (status or "unknown"))

    if wired and executions:
        return ("wired", "reached from %s and running: %d execution(s) seen."
                % (", ".join(sorted(set(wired))), executions))

    if wired:
        return ("wired-idle",
                "attached to %s but no executions in the page read. Wired and "
                "untested, or wired to a line nobody calls."
                % ", ".join(sorted(set(wired))))

    if executions:
        return ("triggered-elsewhere",
                "no number points at it, but %d execution(s) exist: started by "
                "the REST Executions API, a Trigger widget in another Flow, or a "
                "Messaging Service inbound request URL." % executions)

    apps = attach.get("via_application") or []
    hint = ("" if not apps else
            " %d number(s) on this account use voice_application_sid, whose URL "
            "this scan does not follow." % len(apps))
    return ("orphan",
            "published, no number's voice_url or sms_url contains this FlowSid, "
            "and no executions. Inbound traffic is still going wherever it went "
            "before.%s" % hint)


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 paged_v2(session, url, key, limit):
    params = {"PageSize": 50}
    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 list_numbers(session, account, limit):
    """The 2010-04-01 API pages with next_page_uri rather than meta, so this
    cannot share the v2 pager above."""
    url = "%s/Accounts/%s/IncomingPhoneNumbers.json" % (BASE, account)
    params = {"PageSize": 1000}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("incoming_phone_numbers", []))
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-flows", type=int, default=200,
                    help="stop paging after this many Studio Flows")
    ap.add_argument("--max-numbers", type=int, default=5000,
                    help="stop paging after this many phone numbers")
    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)

    flows = paged_v2(session, "%s/Flows" % STUDIO, "flows", args.max_flows)
    if not flows:
        log.info("no Studio Flows on this account")
        return 0

    numbers = list_numbers(session, account, args.max_numbers)
    log.info("%d flow(s), %d number(s) read", len(flows), len(numbers))

    bad = 0
    for flow in flows:
        sid = flow.get("sid")
        attach = attachments(sid, numbers)
        executions = 0
        if not (attach["voice"] or attach["sms"]):
            executions = len(paged_v2(session, "%s/Flows/%s/Executions"
                                      % (STUDIO, sid), "executions", 1))

        state, detail = verdict(flow, attach, executions)
        line = "%-20s %s (%s)  %s" % (state, sid, flow.get("friendly_name", "?"),
                                      detail)
        if state in ("wired", "triggered-elsewhere"):
            log.info(line)
            continue
        if state == "wired-idle":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if state == "unpublished":
            log.warning("  repair: publish the Flow, then attach a number to it.")
            continue
        log.warning("  repair: update %s/Accounts/%s/IncomingPhoneNumbers/{PNSid}"
                    ".json with SmsUrl=https://webhooks.twilio.com/v1/Accounts/%s/"
                    "Flows/%s and SmsMethod=POST (or the VoiceUrl equivalent), or "
                    "assign the number in Console -> Studio -> the Flow.",
                    BASE, account, account, sid)

    log.info("%d published flow(s), %d with no entry point at all",
             sum(1 for f in flows if str(f.get("status") or "") == "published"), bad)
    return 1 if bad else 0


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

/**
 * Report published Twilio Studio Flows that no phone number points at.
 *
 * 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 STUDIO = 'https://studio.twilio.com/v2';

/**
 * Find the numbers whose voice or SMS webhook runs this Flow. Pure.
 *
 * The attachment is a URL, not a reference. Matched as a substring, because the
 * URL can carry a query string and reconstructing the exact expected string is
 * how a scan reports every Flow as unwired. Numbers with voice_application_sid
 * are collected separately: their voice_url is ignored at runtime.
 */
export function attachments(flowSid, numbers) {
  const out = { voice: [], sms: [], via_application: [] };
  if (!flowSid) return out;
  for (const n of numbers ?? []) {
    const label = n.phone_number || n.sid || '?';
    if (String(n.voice_url ?? '').includes(flowSid)) out.voice.push(label);
    if (String(n.sms_url ?? '').includes(flowSid)) out.sms.push(label);
    if (String(n.voice_application_sid ?? '').trim()) out.via_application.push(label);
  }
  return out;
}

/**
 * Classify one published Flow's entry point. Pure, so the difference between
 * "nothing can reach it" and "something reaches it from elsewhere" is written
 * down rather than inferred. Returns [state, detail].
 */
export function verdict(flow, attach = { voice: [], sms: [], via_application: [] },
                        executions = 0) {
  const status = String(flow.status ?? '').toLowerCase();
  const wired = [...(attach.voice ?? []), ...(attach.sms ?? [])];
  const runs = Number(executions ?? 0);

  if (status !== 'published') {
    return ['unpublished',
      `status is ${status || 'unknown'}, so there is no published definition for a ` +
      'number to run. Publish first; wiring a draft changes nothing.'];
  }

  const named = [...new Set(wired)].sort().join(', ');

  if (wired.length && runs) {
    return ['wired', `reached from ${named} and running: ${runs} execution(s) seen.`];
  }

  if (wired.length) {
    return ['wired-idle',
      `attached to ${named} but no executions in the page read. Wired and untested, ` +
      'or wired to a line nobody calls.'];
  }

  if (runs) {
    return ['triggered-elsewhere',
      `no number points at it, but ${runs} execution(s) exist: started by the REST ` +
      'Executions API, a Trigger widget in another Flow, or a Messaging Service ' +
      'inbound request URL.'];
  }

  const apps = attach.via_application ?? [];
  const hint = apps.length
    ? ` ${apps.length} number(s) on this account use voice_application_sid, whose ` +
      'URL this scan does not follow.'
    : '';
  return ['orphan',
    'published, no number\'s voice_url or sms_url contains this FlowSid, and no ' +
    `executions. Inbound traffic is still going wherever it went before.${hint}`];
}

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 pagedV2(auth, url, key, limit = 200) {
  let next = url;
  let params = { PageSize: 50 };
  const out = [];
  while (next && out.length < limit) {
    const page = await get(auth, next, params);
    out.push(...(page[key] ?? []));
    next = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

async function listNumbers(auth, account, limit = 5000) {
  let url = `${BASE}/Accounts/${account}/IncomingPhoneNumbers.json`;
  let params = { PageSize: 1000 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.incoming_phone_numbers ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  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 flows = await pagedV2(auth, `${STUDIO}/Flows`, 'flows');
  if (flows.length === 0) {
    console.log('no Studio Flows on this account');
    return;
  }

  const numbers = await listNumbers(auth, account);
  console.log(`${flows.length} flow(s), ${numbers.length} number(s) read`);

  let bad = 0;
  for (const flow of flows) {
    const attach = attachments(flow.sid, numbers);
    let executions = 0;
    if (!attach.voice.length && !attach.sms.length) {
      const page = await pagedV2(auth, `${STUDIO}/Flows/${flow.sid}/Executions`,
                                 'executions', 1);
      executions = page.length;
    }
    const [state, detail] = verdict(flow, attach, executions);
    const line = `${state.padEnd(20)} ${flow.sid} (${flow.friendly_name ?? '?'})  ${detail}`;
    if (state === 'wired' || state === 'triggered-elsewhere' || state === 'wired-idle') {
      console.log(line);
      continue;
    }
    bad += 1;
    console.warn(line);
    if (state === 'unpublished') {
      console.warn('  repair: publish the Flow, then attach a number to it.');
      continue;
    }
    console.warn(`  repair: update ${BASE}/Accounts/${account}/IncomingPhoneNumbers/` +
                 '{PNSid}.json with SmsUrl=https://webhooks.twilio.com/v1/Accounts/' +
                 `${account}/Flows/${flow.sid} and SmsMethod=POST (or the VoiceUrl ` +
                 'equivalent), or assign the number in Console -> Studio -> the Flow.');
  }

  const published = flows.filter((f) => String(f.status ?? '') === 'published').length;
  console.log(`${published} published flow(s), ${bad} with no entry point at all`);
  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_studio_wiring_audit import attachments, verdict

FLOW = "FW11111111111111111111111111111111"
HOOK = "https://webhooks.twilio.com/v1/Accounts/ACxxx/Flows/" + FLOW


def number(phone="+15550001111", **fields):
    row = {"sid": "PN1", "phone_number": phone, "voice_url": "", "sms_url": "",
           "voice_application_sid": ""}
    row.update(fields)
    return row


def test_a_number_whose_sms_url_is_the_studio_webhook_counts():
    attach = attachments(FLOW, [number(sms_url=HOOK)])
    assert attach["sms"] == ["+15550001111"]
    assert attach["voice"] == []


def test_a_query_string_on_the_webhook_still_matches():
    # Matched as a substring on purpose: an equality test against a rebuilt URL
    # reports every Flow on the account as unwired.
    attach = attachments(FLOW, [number(voice_url=HOOK + "?lang=fr")])
    assert attach["voice"] == ["+15550001111"]


def test_a_different_flow_sid_does_not_match():
    other = "FW22222222222222222222222222222222"
    assert attachments(other, [number(sms_url=HOOK)]) == {
        "voice": [], "sms": [], "via_application": []}


def test_numbers_on_an_application_sid_are_recorded_as_unanswerable():
    attach = attachments(FLOW, [number(voice_application_sid="AP1")])
    assert attach["via_application"] == ["+15550001111"]
    assert attach["voice"] == []


def test_a_wired_flow_with_traffic_is_healthy():
    state, detail = verdict({"status": "published"},
                            {"voice": [], "sms": ["+15550001111"],
                             "via_application": []}, 12)
    assert state == "wired"
    assert "12 execution(s)" in detail


def test_executions_with_no_number_is_not_an_orphan():
    state, detail = verdict({"status": "published"}, None, 40)
    assert state == "triggered-elsewhere"
    assert "REST Executions API" in detail


def test_no_number_and_no_executions_is_the_finding():
    state, detail = verdict({"status": "published"},
                            {"voice": [], "sms": [], "via_application": ["+15550002222"]},
                            0)
    assert state == "orphan"
    assert "voice_application_sid" in detail


def test_a_draft_flow_is_a_different_problem():
    state, detail = verdict({"status": "draft"}, None, 0)
    assert state == "unpublished"
    assert "Publish first" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { attachments, verdict } from './twilio-studio-wiring-audit.mjs';

const FLOW = 'FW11111111111111111111111111111111';
const HOOK = `https://webhooks.twilio.com/v1/Accounts/ACxxx/Flows/${FLOW}`;

const number = (fields = {}) => ({
  sid: 'PN1', phone_number: '+15550001111', voice_url: '', sms_url: '',
  voice_application_sid: '', ...fields,
});

test('a number whose sms_url is the Studio webhook counts', () => {
  const attach = attachments(FLOW, [number({ sms_url: HOOK })]);
  assert.deepEqual(attach.sms, ['+15550001111']);
  assert.deepEqual(attach.voice, []);
});

test('a query string on the webhook still matches', () => {
  const attach = attachments(FLOW, [number({ voice_url: `${HOOK}?lang=fr` })]);
  assert.deepEqual(attach.voice, ['+15550001111']);
});

test('a different flow sid does not match', () => {
  const other = 'FW22222222222222222222222222222222';
  assert.deepEqual(attachments(other, [number({ sms_url: HOOK })]),
    { voice: [], sms: [], via_application: [] });
});

test('numbers on an application sid are recorded as unanswerable', () => {
  const attach = attachments(FLOW, [number({ voice_application_sid: 'AP1' })]);
  assert.deepEqual(attach.via_application, ['+15550001111']);
  assert.deepEqual(attach.voice, []);
});

test('a wired flow with traffic is healthy', () => {
  const [state, detail] = verdict({ status: 'published' },
    { voice: [], sms: ['+15550001111'], via_application: [] }, 12);
  assert.equal(state, 'wired');
  assert.match(detail, /12 execution\(s\)/);
});

test('executions with no number is not an orphan', () => {
  const [state, detail] = verdict({ status: 'published' }, undefined, 40);
  assert.equal(state, 'triggered-elsewhere');
  assert.match(detail, /REST Executions API/);
});

test('no number and no executions is the finding', () => {
  const [state, detail] = verdict({ status: 'published' },
    { voice: [], sms: [], via_application: ['+15550002222'] }, 0);
  assert.equal(state, 'orphan');
  assert.match(detail, /voice_application_sid/);
});

test('a draft flow is a different problem', () => {
  const [state, detail] = verdict({ status: 'draft' }, undefined, 0);
  assert.equal(state, 'unpublished');
  assert.match(detail, /Publish first/);
});