a2p campaign is FAILED and errors[] names the rejected field

"""Report A2P 10DLC campaigns that failed vetting, and name the field that did it.

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

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

# The 308xx/309xx codes that turn up in errors[] on a FAILED campaign, split by
# what actually clears them. A report that only says FAILED sends people round
# the same three week loop; the split is the entire point of this script.
EDITABLE = {
    "30886": ("description", "the use case description is too vague"),
    "30890": ("help_message", "the help message names no brand or support contact"),
    "30892": ("message_samples", "a public URL shortener appears in the samples"),
    "30893": ("message_samples", "the samples do not match the stated use case"),
    "30895": ("direct_lending", "direct lending is not declared"),
    "30909": ("message_flow", "the message flow or call to action is incomplete"),
}
UPSTREAM = {
    "30898": ("brand", "the EIN is already attached to too many brands"),
}
STRUCTURAL = {
    "30883": ("content", "content violation"),
    "30884": ("content", "spam risk"),
    "30885": ("content", "fraud or phishing risk"),
}


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

    The campaign resource spells the key error_code and the brand resource
    spells it code. Reading both is cheaper than being wrong on one of them, and
    normalising to a string means the tables above can be keyed on one type.
    """
    for k in ("error_code", "code"):
        v = err.get(k)
        if v not in (None, ""):
            return str(v)
    return ""


def classify_error(err):
    """Sort one errors[] entry by what will clear it. Pure.

    Returns (bucket, field, why); bucket is editable, upstream, structural or
    unknown.
    """
    code = error_code(err)
    for bucket, table in (("editable", EDITABLE), ("upstream", UPSTREAM),
                          ("structural", STRUCTURAL)):
        if code in table:
            field, why = table[code]
            return (bucket, field, "%s: %s" % (code, why))
    return ("unknown", "",
            "%s: %s" % (code or "no code",
                        err.get("description") or "no description"))


def named_fields(errors):
    """Every campaign attribute the errors point at, in order, without repeats.

    Prefers what the API said in `fields` and falls back to the code table, so a
    code this script has never seen still reports whatever the reviewer named.
    """
    out = []
    for err in errors:
        fields = [str(f).strip() for f in (err.get("fields") or []) if str(f).strip()]
        if not fields:
            _bucket, field, _why = classify_error(err)
            fields = [field] if field else []
        for f in fields:
            if f not in out:
                out.append(f)
    return out


def verdict(campaign):
    """Classify one UsAppToPerson campaign. Pure, so the code table can be
    tested without a network.

    Returns (state, detail).
    """
    if not campaign:
        return ("no-campaign",
                "no A2P campaign on this Messaging Service at all.")

    status = str(campaign.get("campaign_status") or "").upper()
    errors = campaign.get("errors") or []
    buckets = [classify_error(e) for e in errors]
    reasons = "; ".join(w for _b, _f, w in buckets)
    fields = ", ".join(named_fields(errors)) or "nothing named"

    if status == "FAILED":
        if not errors:
            return ("failed-unexplained",
                    "campaign_status is FAILED and errors[] is empty. Nothing "
                    "else in the API explains the rejection, so a resubmission "
                    "now is a guess.")
        if any(b == "structural" for b, _f, _w in buckets):
            return ("failed-structural",
                    "FAILED on a content rejection that editing will not clear "
                    "(%s)." % reasons)
        if any(b == "upstream" for b, _f, _w in buckets):
            return ("failed-at-the-brand",
                    "FAILED on a brand level code (%s). Editing the campaign "
                    "changes nothing until the brand is fixed." % reasons)
        return ("failed-editable",
                "FAILED on %s. Edit %s and resubmit the same campaign."
                % (reasons, fields))

    if status == "SUSPENDED":
        return ("suspended",
                "campaign_status is SUSPENDED, which sends exactly like FAILED. "
                "Check the brand above it before touching the campaign.")

    if status in ("PENDING", "IN_PROGRESS"):
        if errors:
            return ("pending-with-errors",
                    "still %s, but errors[] is already populated (%s): the "
                    "vetting result has arrived and the status has not caught "
                    "up." % (status, reasons))
        return ("pending",
                "still %s: not live, not failed, nothing to edit yet." % status)

    if status == "VERIFIED":
        return ("verified",
                "campaign %s is VERIFIED" % (campaign.get("sid") or "?"))

    return ("unknown-status",
            "campaign_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_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:
        campaigns = list_v1(session,
                            "%s/Services/%s/Compliance/Usa2p" % (MSG, svc["sid"]),
                            "compliance")
        campaign = campaigns[0] if campaigns else None
        state, detail = verdict(campaign)
        name = svc.get("friendly_name") or svc["sid"]
        line = "%-19s %s  %s" % (state, name, detail)
        if state in ("verified", "pending"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        for err in (campaign or {}).get("errors") or []:
            if err.get("url"):
                log.warning("  %s -> %s", error_code(err), err["url"])
        if state == "failed-editable":
            log.warning("  repair: POST %s/Services/%s/Compliance/Usa2p/%s with the "
                        "corrected Description, MessageFlow, MessageSamples or "
                        "HelpMessage", MSG, svc["sid"], campaign.get("sid", "QE..."))
        elif state == "failed-at-the-brand":
            log.warning("  repair: fix the brand first; the campaign edit will not "
                        "take while the brand carries the same error")
        elif state == "failed-structural":
            log.warning("  repair: none by API. The content itself was rejected, so "
                        "the use case has to change before resubmitting")

    log.info("%d service(s), %d with a failed campaign", len(services), bad)
    return 1 if bad else 0


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

/**
 * Report A2P 10DLC campaigns that failed vetting, and name the field that did it.
 *
 * 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';

// The 308xx/309xx codes that turn up in errors[] on a FAILED campaign, split by
// what actually clears them.
const EDITABLE = {
  30886: ['description', 'the use case description is too vague'],
  30890: ['help_message', 'the help message names no brand or support contact'],
  30892: ['message_samples', 'a public URL shortener appears in the samples'],
  30893: ['message_samples', 'the samples do not match the stated use case'],
  30895: ['direct_lending', 'direct lending is not declared'],
  30909: ['message_flow', 'the message flow or call to action is incomplete'],
};
const UPSTREAM = {
  30898: ['brand', 'the EIN is already attached to too many brands'],
};
const STRUCTURAL = {
  30883: ['content', 'content violation'],
  30884: ['content', 'spam risk'],
  30885: ['content', 'fraud or phishing risk'],
};

/**
 * Read the code off one errors[] entry, as a string. The campaign resource
 * spells the key error_code and the brand resource spells it 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 '';
}

/**
 * Sort one errors[] entry by what will clear it. Pure. Returns
 * [bucket, field, why] with bucket editable, upstream, structural or unknown.
 */
export function classifyError(err) {
  const code = errorCode(err);
  for (const [bucket, table] of [['editable', EDITABLE], ['upstream', UPSTREAM],
                                 ['structural', STRUCTURAL]]) {
    if (Object.prototype.hasOwnProperty.call(table, code)) {
      const [field, why] = table[code];
      return [bucket, field, `${code}: ${why}`];
    }
  }
  return ['unknown', '',
          `${code || 'no code'}: ${err.description ?? 'no description'}`];
}

/** Every campaign attribute the errors point at, in order, without repeats. */
export function namedFields(errors) {
  const out = [];
  for (const err of errors) {
    let fields = (err.fields ?? []).map((f) => String(f).trim()).filter(Boolean);
    if (fields.length === 0) {
      const [, field] = classifyError(err);
      fields = field ? [field] : [];
    }
    for (const f of fields) if (!out.includes(f)) out.push(f);
  }
  return out;
}

/**
 * Classify one UsAppToPerson campaign. Pure, so the code table can be tested
 * without a network. Returns [state, detail].
 */
export function verdict(campaign) {
  if (!campaign) {
    return ['no-campaign', 'no A2P campaign on this Messaging Service at all.'];
  }

  const status = String(campaign.campaign_status ?? '').toUpperCase();
  const errors = campaign.errors ?? [];
  const buckets = errors.map(classifyError);
  const reasons = buckets.map(([, , why]) => why).join('; ');
  const fields = namedFields(errors).join(', ') || 'nothing named';

  if (status === 'FAILED') {
    if (errors.length === 0) {
      return ['failed-unexplained',
        'campaign_status is FAILED and errors[] is empty. Nothing else in the ' +
        'API explains the rejection, so a resubmission now is a guess.'];
    }
    if (buckets.some(([b]) => b === 'structural')) {
      return ['failed-structural',
        `FAILED on a content rejection that editing will not clear (${reasons}).`];
    }
    if (buckets.some(([b]) => b === 'upstream')) {
      return ['failed-at-the-brand',
        `FAILED on a brand level code (${reasons}). Editing the campaign ` +
        'changes nothing until the brand is fixed.'];
    }
    return ['failed-editable',
      `FAILED on ${reasons}. Edit ${fields} and resubmit the same campaign.`];
  }

  if (status === 'SUSPENDED') {
    return ['suspended',
      'campaign_status is SUSPENDED, which sends exactly like FAILED. Check ' +
      'the brand above it before touching the campaign.'];
  }

  if (status === 'PENDING' || status === 'IN_PROGRESS') {
    if (errors.length) {
      return ['pending-with-errors',
        `still ${status}, but errors[] is already populated (${reasons}): the ` +
        'vetting result has arrived and the status has not caught up.'];
    }
    return ['pending', `still ${status}: not live, not failed, nothing to edit yet.`];
  }

  if (status === 'VERIFIED') {
    return ['verified', `campaign ${campaign.sid ?? '?'} is VERIFIED`];
  }

  return ['unknown-status',
    `campaign_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();
}

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 campaign = campaigns[0] ?? null;
    const [state, detail] = verdict(campaign);
    const name = svc.friendly_name ?? svc.sid;
    const line = `${state.padEnd(19)} ${name}  ${detail}`;
    if (state === 'verified' || state === 'pending') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    for (const err of campaign?.errors ?? []) {
      if (err.url) console.warn(`  ${errorCode(err)} -> ${err.url}`);
    }
    if (state === 'failed-editable') {
      console.warn(`  repair: POST ${MSG}/Services/${svc.sid}/Compliance/Usa2p/` +
                   `${campaign.sid ?? 'QE...'} with the corrected Description, ` +
                   'MessageFlow, MessageSamples or HelpMessage');
    } else if (state === 'failed-at-the-brand') {
      console.warn('  repair: fix the brand first; the campaign edit will not take ' +
                   'while the brand carries the same error');
    } else if (state === 'failed-structural') {
      console.warn('  repair: none by API. The content itself was rejected, so the ' +
                   'use case has to change before resubmitting');
    }
  }

  console.log(`${services.length} service(s), ${bad} with a failed campaign`);
  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_campaign_vetting_audit import (classify_error, named_fields,
                                                  verdict)

FAILED = {"sid": "QE0123456789", "campaign_status": "FAILED"}


def test_failed_on_an_editable_code_names_the_field_to_change():
    state, detail = verdict(dict(FAILED, errors=[{"error_code": 30893,
                                                  "fields": ["message_samples"]}]))
    assert state == "failed-editable"
    assert "message_samples" in detail


def test_content_rejection_is_not_an_edit():
    # 30884 is a spam risk judgement. Rewriting the description does not clear it,
    # and reporting it in the same bucket costs weeks.
    state, detail = verdict(dict(FAILED, errors=[{"error_code": "30884"}]))
    assert state == "failed-structural"
    assert "will not clear" in detail


def test_ein_code_points_at_the_brand_not_the_campaign():
    state, _ = verdict(dict(FAILED, errors=[{"error_code": 30898}]))
    assert state == "failed-at-the-brand"


def test_failed_with_an_empty_errors_array_is_its_own_state():
    state, detail = verdict(dict(FAILED, errors=[]))
    assert state == "failed-unexplained"
    assert "guess" in detail


def test_an_error_object_spelled_code_is_still_read():
    # The campaign resource says error_code and the brand resource says code.
    bucket, field, _why = classify_error({"code": "30886"})
    assert (bucket, field) == ("editable", "description")


def test_fields_from_the_api_win_over_the_table():
    assert named_fields([{"error_code": 30886, "fields": ["message_flow"]}]) == \
        ["message_flow"]


def test_in_progress_with_errors_is_not_reported_as_waiting():
    state, _ = verdict({"campaign_status": "IN_PROGRESS",
                        "errors": [{"error_code": 30909}]})
    assert state == "pending-with-errors"


def test_verified_campaign_is_clean():
    state, detail = verdict({"campaign_status": "VERIFIED", "sid": "QE0123456789"})
    assert state == "verified"
    assert "QE0123456789" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { classifyError, namedFields, verdict }
  from './twilio-a2p-campaign-vetting-audit.mjs';

const FAILED = { sid: 'QE0123456789', campaign_status: 'FAILED' };

test('failed on an editable code names the field to change', () => {
  const [state, detail] = verdict({ ...FAILED,
    errors: [{ error_code: 30893, fields: ['message_samples'] }] });
  assert.equal(state, 'failed-editable');
  assert.match(detail, /message_samples/);
});

test('content rejection is not an edit', () => {
  const [state, detail] = verdict({ ...FAILED, errors: [{ error_code: '30884' }] });
  assert.equal(state, 'failed-structural');
  assert.match(detail, /will not clear/);
});

test('ein code points at the brand, not the campaign', () => {
  assert.equal(verdict({ ...FAILED, errors: [{ error_code: 30898 }] })[0],
               'failed-at-the-brand');
});

test('failed with an empty errors array is its own state', () => {
  const [state, detail] = verdict({ ...FAILED, errors: [] });
  assert.equal(state, 'failed-unexplained');
  assert.match(detail, /guess/);
});

test('an error object spelled code is still read', () => {
  const [bucket, field] = classifyError({ code: '30886' });
  assert.deepEqual([bucket, field], ['editable', 'description']);
});

test('fields from the api win over the table', () => {
  assert.deepEqual(namedFields([{ error_code: 30886, fields: ['message_flow'] }]),
                   ['message_flow']);
});

test('in progress with errors is not reported as waiting', () => {
  assert.equal(
    verdict({ campaign_status: 'IN_PROGRESS', errors: [{ error_code: 30909 }] })[0],
    'pending-with-errors');
});

test('verified campaign is clean', () => {
  const [state, detail] = verdict({ campaign_status: 'VERIFIED', sid: 'QE0123456789' });
  assert.equal(state, 'verified');
  assert.match(detail, /QE0123456789/);
});