a Messaging Service with no A2P campaign fails US sends

"""Report Messaging Services that cannot send to US numbers under A2P 10DLC.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The registration 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_a2p_registration_audit")

MSG = "https://messaging.twilio.com/v1"

TOLL_FREE = ("800", "833", "844", "855", "866", "877", "888")


def us_long_codes(pool):
    """Count the senders 10DLC registration actually governs.

    Pure. Toll-free numbers are verified separately and short codes are not
    10DLC at all, so counting every +1 in the pool overstates the exposure.
    """
    out = []
    for n in pool:
        number = str(n.get("phone_number") or "")
        if not number.startswith("+1") or len(number) != 12:
            continue
        if number[2:5] in TOLL_FREE:
            continue
        out.append(number)
    return out


def verdict(service, campaigns, us_senders):
    """Classify one Messaging Service's A2P standing. Pure, so the states can be
    tested without a network.

    `campaigns` is the list from Compliance/Usa2p; `us_senders` is the count of
    US long codes in its pool. Returns (state, detail).
    """
    registered = bool(service.get("us_app_to_person_registered"))
    campaign = campaigns[0] if campaigns else None

    if campaign is None:
        if registered:
            return ("inconsistent",
                    "us_app_to_person_registered is true but Compliance/Usa2p "
                    "returned no campaign. Trust the subresource, not the flag.")
        if us_senders:
            return ("blocked",
                    "no A2P campaign and %d US long code(s) in the pool: every "
                    "US send through this service returns 30034." % us_senders)
        return ("unregistered",
                "no A2P campaign. No US long codes in the pool yet, so nothing "
                "is failing; register before one is added.")

    status = str(campaign.get("campaign_status") or "").upper()
    if status == "VERIFIED":
        if not registered:
            return ("inconsistent",
                    "campaign is VERIFIED but us_app_to_person_registered is "
                    "false. Trust the subresource, not the flag.")
        return ("registered", "campaign %s is VERIFIED" % campaign.get("sid", "?"))

    return ("campaign-%s" % (status.lower() or "unknown"),
            "a campaign exists but its status is %s, which sends exactly like "
            "no campaign at all (%d US long code(s) affected)."
            % (status or "unset", us_senders))


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 list_v1(session, url, key, limit=1000):
    """Page a messaging.twilio.com list. meta.next_page_url is absolute."""
    out = []
    while url and len(out) < limit:
        page = get(session, url, PageSize=50)
        out.extend(page.get(key, []))
        url = (page.get("meta") or {}).get("next_page_url")
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-services", type=int, default=200)
    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)

    services = list_v1(session, MSG + "/Services", "services", args.max_services)
    if not services:
        log.info("no Messaging Services on this account")
        return 0

    bad = 0
    for svc in services:
        sid = svc["sid"]
        campaigns = list_v1(session, "%s/Services/%s/Compliance/Usa2p" % (MSG, sid),
                            "compliance")
        pool = list_v1(session, "%s/Services/%s/PhoneNumbers" % (MSG, sid),
                       "phone_numbers")
        state, detail = verdict(svc, campaigns, len(us_long_codes(pool)))

        line = "%-22s %s  %s" % (state, svc.get("friendly_name", sid), detail)
        if state == "registered":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if state in ("blocked", "unregistered"):
            log.warning("  repair: POST %s/Services/%s/Compliance/Usa2p with "
                        "BrandRegistrationSid, Description, MessageFlow, "
                        "MessageSamples, UsAppToPersonUsecase, HasEmbeddedLinks, "
                        "HasEmbeddedPhone", MSG, sid)

    log.info("%d service(s), %d unable to send to US numbers", len(services), bad)
    return 1 if bad else 0


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

/**
 * Report Messaging Services that cannot send to US numbers under A2P 10DLC.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The registration is printed, never
 * performed.
 */
const MSG = 'https://messaging.twilio.com/v1';

const TOLL_FREE = ['800', '833', '844', '855', '866', '877', '888'];

/**
 * Count the senders 10DLC registration actually governs. Pure: toll-free
 * numbers are verified separately and short codes are not 10DLC at all.
 */
export function usLongCodes(pool) {
  return pool
    .map((n) => String(n.phone_number ?? ''))
    .filter((n) => n.startsWith('+1') && n.length === 12 && !TOLL_FREE.includes(n.slice(2, 5)));
}

/**
 * Classify one Messaging Service's A2P standing. Pure, so the states can be
 * tested without a network. Returns [state, detail].
 */
export function verdict(service, campaigns, usSenders) {
  const registered = Boolean(service.us_app_to_person_registered);
  const campaign = campaigns && campaigns.length ? campaigns[0] : null;

  if (campaign === null) {
    if (registered) {
      return ['inconsistent',
        'us_app_to_person_registered is true but Compliance/Usa2p returned no ' +
        'campaign. Trust the subresource, not the flag.'];
    }
    if (usSenders) {
      return ['blocked',
        `no A2P campaign and ${usSenders} US long code(s) in the pool: every US ` +
        'send through this service returns 30034.'];
    }
    return ['unregistered',
      'no A2P campaign. No US long codes in the pool yet, so nothing is ' +
      'failing; register before one is added.'];
  }

  const status = String(campaign.campaign_status ?? '').toUpperCase();
  if (status === 'VERIFIED') {
    if (!registered) {
      return ['inconsistent',
        'campaign is VERIFIED but us_app_to_person_registered is false. Trust ' +
        'the subresource, not the flag.'];
    }
    return ['registered', `campaign ${campaign.sid ?? '?'} is VERIFIED`];
  }

  return [`campaign-${status.toLowerCase() || 'unknown'}`,
    `a campaign exists but its status is ${status || 'unset'}, which sends ` +
    `exactly like no campaign at all (${usSenders} US long code(s) affected).`];
}

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 listV1(auth, url, key, limit = 1000) {
  const out = [];
  let next = url;
  while (next && out.length < limit) {
    const page = await get(auth, next, { PageSize: 50 });
    out.push(...(page[key] ?? []));
    next = page.meta?.next_page_url ?? null;
  }
  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 services = await listV1(auth, `${MSG}/Services`, 'services');
  if (services.length === 0) {
    console.log('no Messaging Services on this account');
    return;
  }

  let bad = 0;
  for (const svc of services) {
    const campaigns = await listV1(auth, `${MSG}/Services/${svc.sid}/Compliance/Usa2p`,
                                   'compliance');
    const pool = await listV1(auth, `${MSG}/Services/${svc.sid}/PhoneNumbers`,
                              'phone_numbers');
    const [state, detail] = verdict(svc, campaigns, usLongCodes(pool).length);

    const line = `${state.padEnd(22)} ${svc.friendly_name ?? svc.sid}  ${detail}`;
    if (state === 'registered') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    if (state === 'blocked' || state === 'unregistered') {
      console.warn(`  repair: POST ${MSG}/Services/${svc.sid}/Compliance/Usa2p with ` +
                   'BrandRegistrationSid, Description, MessageFlow, MessageSamples, ' +
                   'UsAppToPersonUsecase, HasEmbeddedLinks, HasEmbeddedPhone');
    }
  }

  console.log(`${services.length} service(s), ${bad} unable to send to US numbers`);
  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_a2p_registration_audit import us_long_codes, verdict

REGISTERED = {"us_app_to_person_registered": True}
UNREGISTERED = {"us_app_to_person_registered": False}
VERIFIED = [{"sid": "QE0123456789", "campaign_status": "VERIFIED"}]


def test_unregistered_with_us_senders_is_an_outage():
    state, detail = verdict(UNREGISTERED, [], 3)
    assert state == "blocked"
    assert "30034" in detail


def test_unregistered_with_no_us_senders_is_not_the_same_finding():
    # Same missing campaign, but nothing is failing yet. Keeping these apart is
    # what makes the report worth reading on a big account.
    state, _ = verdict(UNREGISTERED, [], 0)
    assert state == "unregistered"


def test_verified_campaign_and_flag_agree():
    state, detail = verdict(REGISTERED, VERIFIED, 3)
    assert state == "registered"
    assert "QE0123456789" in detail


def test_campaign_in_progress_sends_like_no_campaign():
    state, detail = verdict(REGISTERED, [{"campaign_status": "IN_PROGRESS"}], 2)
    assert state == "campaign-in_progress"
    assert "no campaign at all" in detail


def test_suspended_campaign_is_not_reported_as_registered():
    state, _ = verdict(REGISTERED, [{"campaign_status": "SUSPENDED"}], 1)
    assert state == "campaign-suspended"


def test_flag_disagreeing_with_the_subresource_is_reported():
    state, _ = verdict(REGISTERED, [], 1)
    assert state == "inconsistent"


def test_toll_free_and_short_codes_are_not_10dlc_senders():
    pool = [{"phone_number": "+15550001111"}, {"phone_number": "+18885551234"},
            {"phone_number": "+447700900123"}, {"phone_number": "12345"}]
    assert us_long_codes(pool) == ["+15550001111"]

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

const REGISTERED = { us_app_to_person_registered: true };
const UNREGISTERED = { us_app_to_person_registered: false };
const VERIFIED = [{ sid: 'QE0123456789', campaign_status: 'VERIFIED' }];

test('unregistered with us senders is an outage', () => {
  const [state, detail] = verdict(UNREGISTERED, [], 3);
  assert.equal(state, 'blocked');
  assert.match(detail, /30034/);
});

test('unregistered with no us senders is not the same finding', () => {
  assert.equal(verdict(UNREGISTERED, [], 0)[0], 'unregistered');
});

test('verified campaign and flag agree', () => {
  const [state, detail] = verdict(REGISTERED, VERIFIED, 3);
  assert.equal(state, 'registered');
  assert.match(detail, /QE0123456789/);
});

test('campaign in progress sends like no campaign', () => {
  const [state, detail] = verdict(REGISTERED, [{ campaign_status: 'IN_PROGRESS' }], 2);
  assert.equal(state, 'campaign-in_progress');
  assert.match(detail, /no campaign at all/);
});

test('suspended campaign is not reported as registered', () => {
  assert.equal(verdict(REGISTERED, [{ campaign_status: 'SUSPENDED' }], 1)[0],
               'campaign-suspended');
});

test('flag disagreeing with the subresource is reported', () => {
  assert.equal(verdict(REGISTERED, [], 1)[0], 'inconsistent');
});

test('toll free and short codes are not 10dlc senders', () => {
  const pool = [{ phone_number: '+15550001111' }, { phone_number: '+18885551234' },
                { phone_number: '+447700900123' }, { phone_number: '12345' }];
  assert.deepEqual(usLongCodes(pool), ['+15550001111']);
});