an A2P brand stuck at FAILED blocks every campaign under it

"""Report A2P 10DLC brands that block every campaign underneath them.

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

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

DELETING = ("DELETION_PENDING", "DELETION_FAILED")
WAITING = ("PENDING", "IN_REVIEW")

# Superseded by errors[]. Read only as a labelled fallback, because an
# integration written against them reports "no reason given" on a brand that
# explained itself in full.
DEPRECATED = ("failure_reason", "brand_feedback")


def error_code(err):
    """Read the code off one errors[] entry, as a string.

    The brand resource spells the key code and the campaign resource spells it
    error_code. Reading both costs one loop and removes a whole class of silent
    misreport.
    """
    for k in ("error_code", "code"):
        v = err.get(k)
        if v not in (None, ""):
            return str(v)
    return ""


def failure_lines(brand):
    """The reasons a brand gives for its state, and where they came from. Pure.

    Returns (source, lines). source is "errors" when errors[] carried them,
    "deprecated" when only the old prose fields did, and "none" when the brand
    offers no explanation at all.
    """
    lines = []
    for err in brand.get("errors") or []:
        fields = ", ".join(str(f).strip() for f in (err.get("fields") or [])
                           if str(f).strip())
        text = "%s: %s" % (error_code(err) or "no code",
                           err.get("description") or "no description")
        lines.append("%s (%s)" % (text, fields) if fields else text)
    if lines:
        return ("errors", lines)

    for key in DEPRECATED:
        value = str(brand.get(key) or "").strip()
        if value:
            lines.append("%s: %s" % (key, value))
    if lines:
        return ("deprecated", lines)

    return ("none", [])


def verdict(brand):
    """Classify one BrandRegistration. Pure, so the states can be tested without
    a network.

    Returns (state, detail).
    """
    status = str(brand.get("status") or "").upper()
    tcr = str(brand.get("tcr_id") or "").strip()
    source, lines = failure_lines(brand)
    reasons = "; ".join(lines)

    if status == "FAILED":
        if source == "errors":
            return ("failed",
                    "brand is FAILED: %s. No campaign can attach while it stays "
                    "here, so every US send is 30034." % reasons)
        if source == "deprecated":
            return ("failed-deprecated-reason",
                    "brand is FAILED and errors[] is empty; the only text "
                    "available is from a deprecated field (%s)." % reasons)
        return ("failed-unexplained",
                "brand is FAILED with an empty errors[] and no legacy text. "
                "Re-fetch before resubmitting: there are only three free "
                "resubmissions and a fourth returns 21724.")

    if status == "SUSPENDED":
        return ("suspended",
                "brand is SUSPENDED, which suspends every campaign under it. "
                "%s" % (reasons or "No reason on the resource; this is a "
                        "support conversation, not an API repair."))

    if status in DELETING:
        return ("deleting",
                "brand is %s: it is on its way out and cannot carry a campaign."
                % status)

    if status in WAITING:
        return ("in-review",
                "brand is %s and tcr_id is %s. Not failed, just not usable yet."
                % (status, tcr or "null"))

    if status == "APPROVED":
        if not tcr:
            return ("approved-no-tcr-id",
                    "status is APPROVED but tcr_id is null, which is what an "
                    "unapproved brand looks like. Report the disagreement "
                    "rather than picking a side.")
        return ("approved", "brand is APPROVED with tcr_id %s" % tcr)

    return ("unknown-status",
            "status is %s, which this script does not recognise."
            % (status or "unset"))


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_brands(session, limit=500):
    """Page the brand list.

    This resource returns its items under `data`, not under a resource-named key
    like the rest of messaging v1. meta.next_page_url is absolute.
    """
    url = MSG + "/a2p/BrandRegistrations"
    out = []
    while url and len(out) < limit:
        page = get(session, url, PageSize=50)
        out.extend(page.get("data", []))
        url = (page.get("meta") or {}).get("next_page_url")
    return out[:limit]


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

    brands = list_brands(session, args.max_brands)
    if not brands:
        log.info("no A2P brand registrations on this account")
        return 0

    bad = 0
    for brand in brands:
        state, detail = verdict(brand)
        sid = brand.get("sid", "?")
        line = "%-24s %s  %s" % (state, sid, detail)
        if state in ("approved", "in-review"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        for err in brand.get("errors") or []:
            if err.get("url"):
                log.warning("  %s -> %s", error_code(err), err["url"])
        if state in ("failed", "failed-deprecated-reason", "failed-unexplained"):
            log.warning("  repair: correct the Customer Profile bundle %s in Trust "
                        "Hub, then POST %s/a2p/BrandRegistrations/%s to resubmit",
                        brand.get("customer_profile_bundle_sid", "BU..."), MSG, sid)
        elif state == "suspended":
            log.warning("  repair: none by API. Resolve the suspension with Twilio "
                        "Support; do not move the traffic to a new brand")

    log.info("%d brand(s), %d blocking campaign registration", len(brands), bad)
    return 1 if bad else 0


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

/**
 * Report A2P 10DLC brands that block every campaign underneath them.
 *
 * 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 MSG = 'https://messaging.twilio.com/v1';

const DELETING = ['DELETION_PENDING', 'DELETION_FAILED'];
const WAITING = ['PENDING', 'IN_REVIEW'];

// Superseded by errors[]. Read only as a labelled fallback.
const DEPRECATED = ['failure_reason', 'brand_feedback'];

/**
 * Read the code off one errors[] entry, as a string. The brand resource spells
 * the key code and the campaign resource spells it error_code.
 */
export function errorCode(err) {
  for (const k of ['error_code', 'code']) {
    const v = err[k];
    if (v !== undefined && v !== null && v !== '') return String(v);
  }
  return '';
}

/**
 * The reasons a brand gives for its state, and where they came from. Pure.
 * Returns [source, lines] with source errors, deprecated or none.
 */
export function failureLines(brand) {
  const lines = [];
  for (const err of brand.errors ?? []) {
    const fields = (err.fields ?? []).map((f) => String(f).trim())
      .filter(Boolean).join(', ');
    const text = `${errorCode(err) || 'no code'}: ${err.description ?? 'no description'}`;
    lines.push(fields ? `${text} (${fields})` : text);
  }
  if (lines.length) return ['errors', lines];

  for (const key of DEPRECATED) {
    const value = String(brand[key] ?? '').trim();
    if (value) lines.push(`${key}: ${value}`);
  }
  if (lines.length) return ['deprecated', lines];

  return ['none', []];
}

/**
 * Classify one BrandRegistration. Pure, so the states can be tested without a
 * network. Returns [state, detail].
 */
export function verdict(brand) {
  const status = String(brand.status ?? '').toUpperCase();
  const tcr = String(brand.tcr_id ?? '').trim();
  const [source, lines] = failureLines(brand);
  const reasons = lines.join('; ');

  if (status === 'FAILED') {
    if (source === 'errors') {
      return ['failed',
        `brand is FAILED: ${reasons}. No campaign can attach while it stays ` +
        'here, so every US send is 30034.'];
    }
    if (source === 'deprecated') {
      return ['failed-deprecated-reason',
        'brand is FAILED and errors[] is empty; the only text available is ' +
        `from a deprecated field (${reasons}).`];
    }
    return ['failed-unexplained',
      'brand is FAILED with an empty errors[] and no legacy text. Re-fetch ' +
      'before resubmitting: there are only three free resubmissions and a ' +
      'fourth returns 21724.'];
  }

  if (status === 'SUSPENDED') {
    return ['suspended',
      'brand is SUSPENDED, which suspends every campaign under it. ' +
      (reasons || 'No reason on the resource; this is a support conversation, ' +
       'not an API repair.')];
  }

  if (DELETING.includes(status)) {
    return ['deleting',
      `brand is ${status}: it is on its way out and cannot carry a campaign.`];
  }

  if (WAITING.includes(status)) {
    return ['in-review',
      `brand is ${status} and tcr_id is ${tcr || 'null'}. Not failed, just not ` +
      'usable yet.'];
  }

  if (status === 'APPROVED') {
    if (!tcr) {
      return ['approved-no-tcr-id',
        'status is APPROVED but tcr_id is null, which is what an unapproved ' +
        'brand looks like. Report the disagreement rather than picking a side.'];
    }
    return ['approved', `brand is APPROVED with tcr_id ${tcr}`];
  }

  return ['unknown-status',
    `status is ${status || 'unset'}, which this script does not recognise.`];
}

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 the brand list. This resource returns its items under `data`, not under
 * a resource-named key like the rest of messaging v1.
 */
export async function listBrands(auth, limit = 500) {
  const out = [];
  let next = `${MSG}/a2p/BrandRegistrations`;
  while (next && out.length < limit) {
    const page = await get(auth, next, { PageSize: 50 });
    out.push(...(page.data ?? []));
    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 brands = await listBrands(auth);
  if (brands.length === 0) {
    console.log('no A2P brand registrations on this account');
    return;
  }

  let bad = 0;
  for (const brand of brands) {
    const [state, detail] = verdict(brand);
    const sid = brand.sid ?? '?';
    const line = `${state.padEnd(24)} ${sid}  ${detail}`;
    if (state === 'approved' || state === 'in-review') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    for (const err of brand.errors ?? []) {
      if (err.url) console.warn(`  ${errorCode(err)} -> ${err.url}`);
    }
    if (state.startsWith('failed')) {
      console.warn(`  repair: correct the Customer Profile bundle ` +
                   `${brand.customer_profile_bundle_sid ?? 'BU...'} in Trust Hub, ` +
                   `then POST ${MSG}/a2p/BrandRegistrations/${sid} to resubmit`);
    } else if (state === 'suspended') {
      console.warn('  repair: none by API. Resolve the suspension with Twilio ' +
                   'Support; do not move the traffic to a new brand');
    }
  }

  console.log(`${brands.length} brand(s), ${bad} blocking campaign registration`);
  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_brand_audit import failure_lines, verdict


def test_failed_brand_reports_the_code_and_the_fields():
    state, detail = verdict({
        "status": "FAILED",
        "errors": [{"code": 30799, "description": "Unable to verify registration "
                    "details", "fields": ["business_registration_identifier"]}],
    })
    assert state == "failed"
    assert "30799" in detail
    assert "business_registration_identifier" in detail


def test_deprecated_prose_is_used_but_labelled():
    # failure_reason and brand_feedback are superseded by errors[]. If they are
    # all that is populated, say so rather than presenting them as the answer.
    state, detail = verdict({"status": "FAILED", "errors": [],
                             "failure_reason": "EIN does not match"})
    assert state == "failed-deprecated-reason"
    assert "deprecated" in detail


def test_errors_win_over_the_deprecated_fields():
    source, lines = failure_lines({"errors": [{"code": "30799"}],
                                   "brand_feedback": "old text"})
    assert source == "errors"
    assert len(lines) == 1


def test_failed_with_nothing_at_all_mentions_the_resubmission_limit():
    state, detail = verdict({"status": "FAILED"})
    assert state == "failed-unexplained"
    assert "21724" in detail


def test_approved_without_a_tcr_id_is_a_disagreement():
    state, _ = verdict({"status": "APPROVED", "tcr_id": None})
    assert state == "approved-no-tcr-id"


def test_approved_with_a_tcr_id_is_clean():
    state, detail = verdict({"status": "APPROVED", "tcr_id": "BRAND1234"})
    assert state == "approved"
    assert "BRAND1234" in detail


def test_suspended_is_not_folded_into_failed():
    state, detail = verdict({"status": "SUSPENDED"})
    assert state == "suspended"
    assert "every campaign" in detail


def test_in_review_is_not_a_finding():
    state, _ = verdict({"status": "IN_REVIEW", "tcr_id": None})
    assert state == "in-review"

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

test('failed brand reports the code and the fields', () => {
  const [state, detail] = verdict({
    status: 'FAILED',
    errors: [{ code: 30799, description: 'Unable to verify registration details',
               fields: ['business_registration_identifier'] }],
  });
  assert.equal(state, 'failed');
  assert.match(detail, /30799/);
  assert.match(detail, /business_registration_identifier/);
});

test('deprecated prose is used but labelled', () => {
  const [state, detail] = verdict({ status: 'FAILED', errors: [],
                                    failure_reason: 'EIN does not match' });
  assert.equal(state, 'failed-deprecated-reason');
  assert.match(detail, /deprecated/);
});

test('errors win over the deprecated fields', () => {
  const [source, lines] = failureLines({ errors: [{ code: '30799' }],
                                         brand_feedback: 'old text' });
  assert.equal(source, 'errors');
  assert.equal(lines.length, 1);
});

test('failed with nothing at all mentions the resubmission limit', () => {
  const [state, detail] = verdict({ status: 'FAILED' });
  assert.equal(state, 'failed-unexplained');
  assert.match(detail, /21724/);
});

test('approved without a tcr id is a disagreement', () => {
  assert.equal(verdict({ status: 'APPROVED', tcr_id: null })[0],
               'approved-no-tcr-id');
});

test('approved with a tcr id is clean', () => {
  const [state, detail] = verdict({ status: 'APPROVED', tcr_id: 'BRAND1234' });
  assert.equal(state, 'approved');
  assert.match(detail, /BRAND1234/);
});

test('suspended is not folded into failed', () => {
  const [state, detail] = verdict({ status: 'SUSPENDED' });
  assert.equal(state, 'suspended');
  assert.match(detail, /every campaign/);
});

test('in review is not a finding', () => {
  assert.equal(verdict({ status: 'IN_REVIEW', tcr_id: null })[0], 'in-review');
});