a suspended campaign returns 30033 and the sends keep coming

"""Date a 10DLC campaign suspension from the Messages list and say what happened next.

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 datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime

import requests

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

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

SUSPENDED = "30033"


def parse_when(value):
    """date_sent is RFC 2822, not ISO 8601. Returns epoch seconds, or None.

    Lenient on purpose. One malformed row should cost one row, not the window,
    and the rows that matter most here are the oldest ones in it.
    """
    if not value:
        return None
    try:
        return parsedate_to_datetime(value).timestamp()
    except (TypeError, ValueError):
        return None


def sender_key(message):
    """What carried this message.

    The Messaging Service wins when both are set, because the campaign is
    attached to the service rather than to the number.
    """
    for k in ("messaging_service_sid", "from"):
        value = message.get(k)
        if value:
            return str(value)
    return "unknown"


def ordered(messages):
    """Oldest first. Rows with no usable date_sent keep their original order at
    the end rather than being dropped, so a bad timestamp cannot hide a 30033."""
    keyed = [(parse_when(m.get("date_sent")), i, m) for i, m in enumerate(messages)]
    dated = sorted(k for k in keyed if k[0] is not None)
    undated = [k for k in keyed if k[0] is None]
    return [m for _w, _i, m in dated] + [m for _w, _i, m in undated]


def is_suspended(message):
    return str(message.get("error_code") or "") == SUSPENDED


def recipients(messages):
    """Distinct to values. Retries turn one blocked customer into three rows, so
    this is the number that belongs in the customer comms."""
    return len({str(m.get("to") or "") for m in messages if m.get("to")})


def verdict(messages):
    """Classify a window by what the sends did after the first 30033. Pure.

    Returns (state, detail). States: clean, rerouted, still-pushing, stopped.
    """
    rows = ordered(messages)
    blocked = [m for m in rows if is_suspended(m)]
    if not blocked:
        return ("clean", "no 30033 in this window.")

    first = next(i for i, m in enumerate(rows) if is_suspended(m))
    after = rows[first + 1:]
    later = [m for m in after if is_suspended(m)]

    partial = ""
    seen_before = None
    if first == 0:
        partial = (" The window opens on a 30033, so the suspension started "
                   "before it: widen --days before reading anything into which "
                   "senders look new.")
    else:
        seen_before = {sender_key(m) for m in rows[:first]}

    if seen_before is not None:
        fresh = []
        for m in after:
            key = sender_key(m)
            if key not in seen_before and not is_suspended(m) and key not in fresh:
                fresh.append(key)
        if fresh:
            return ("rerouted",
                    "%d x 30033 over %d recipient(s), and then %s started "
                    "carrying traffic that had never used it before. Moving "
                    "suspended traffic to another sender is the response that "
                    "escalates to account termination."
                    % (len(blocked), recipients(blocked), ", ".join(fresh)))

    if later:
        return ("still-pushing",
                "%d x 30033 over %d recipient(s), %d of them after the first. "
                "The producer has not been told to stop and every one of those "
                "is a send that was refused.%s"
                % (len(blocked), recipients(blocked), len(later), partial))

    return ("stopped",
            "%d x 30033 over %d recipient(s), and nothing refused since. The "
            "sending stopped; the suspension is open until Support clears it.%s"
            % (len(blocked), recipients(blocked), partial))


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_messages(session, account, since, limit):
    """Page Messages.json. There is no Status or ErrorCode filter on this
    resource, so the window and the page cap are the only ways to bound it."""
    url = "%s/Accounts/%s/Messages.json" % (BASE, account)
    params = {"PageSize": 1000, "DateSent>=": since}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("messages", []))
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


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("--days", type=int, default=14,
                    help="how far back to read the Messages list")
    ap.add_argument("--max-messages", type=int, default=20000,
                    help="stop paging after this many messages")
    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)

    since = (datetime.now(timezone.utc)
             - timedelta(days=args.days)).strftime("%Y-%m-%d")
    messages = list_messages(session, account, since, args.max_messages)
    if not messages:
        log.info("no messages sent since %s", since)
        return 0

    state, detail = verdict(messages)
    line = "%-14s %s  %d message(s) since %s" % (state, detail, len(messages), since)
    if state == "clean":
        log.info(line)
        return 0
    log.warning(line)

    blocked = [m for m in messages if is_suspended(m)]
    services = set()
    for sender in sorted({sender_key(m) for m in blocked}):
        count = len([m for m in blocked if sender_key(m) == sender])
        log.warning("  %s  %d x 30033", sender, count)
        if sender.startswith("MG"):
            services.add(sender)

    for service in sorted(services):
        campaigns = list_v1(session, "%s/Services/%s/Compliance/Usa2p" % (MSG, service),
                            "compliance")
        status = (campaigns[0].get("campaign_status") if campaigns else None)
        log.warning("  %s  campaign_status=%s", service, status or "no campaign")

    log.warning("  repair: none by API. Stop the producer, remediate the traffic "
                "named in the suspension notice and reply to Twilio Support with "
                "evidence. Check the brand above the campaign before assuming the "
                "decision was made at the campaign.")
    if state == "rerouted":
        log.warning("  repair: undo the reroute first. Sending the same traffic "
                    "from another sender escalates a campaign suspension to the "
                    "account.")
    return 1


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

/**
 * Date a 10DLC campaign suspension from the Messages list and say what happened next.
 *
 * 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 MSG = 'https://messaging.twilio.com/v1';

const SUSPENDED = '30033';

/**
 * date_sent is RFC 2822, not ISO 8601. Returns epoch seconds, or null. Lenient
 * on purpose: one malformed row should cost one row, not the window.
 */
export function parseWhen(value) {
  if (!value) return null;
  const ms = Date.parse(value);
  return Number.isNaN(ms) ? null : ms / 1000;
}

/**
 * What carried this message. The Messaging Service wins when both are set,
 * because the campaign is attached to the service rather than to the number.
 */
export function senderKey(message) {
  for (const k of ['messaging_service_sid', 'from']) {
    if (message[k]) return String(message[k]);
  }
  return 'unknown';
}

/** Oldest first. Rows with no usable date_sent keep their order at the end. */
export function ordered(messages) {
  const keyed = messages.map((m, i) => [parseWhen(m.date_sent), i, m]);
  const dated = keyed.filter(([w]) => w !== null)
    .sort((a, b) => (a[0] - b[0]) || (a[1] - b[1]));
  const undated = keyed.filter(([w]) => w === null);
  return [...dated, ...undated].map(([, , m]) => m);
}

export function isSuspended(message) {
  return String(message.error_code ?? '') === SUSPENDED;
}

/** Distinct to values: retries turn one blocked customer into three rows. */
export function recipients(messages) {
  return new Set(messages.filter((m) => m.to).map((m) => String(m.to))).size;
}

/**
 * Classify a window by what the sends did after the first 30033. Pure. Returns
 * [state, detail] with state clean, rerouted, still-pushing or stopped.
 */
export function verdict(messages) {
  const rows = ordered(messages);
  const blocked = rows.filter(isSuspended);
  if (blocked.length === 0) return ['clean', 'no 30033 in this window.'];

  const first = rows.findIndex(isSuspended);
  const after = rows.slice(first + 1);
  const later = after.filter(isSuspended);

  let partial = '';
  let seenBefore = null;
  if (first === 0) {
    partial = ' The window opens on a 30033, so the suspension started before ' +
      'it: widen --days before reading anything into which senders look new.';
  } else {
    seenBefore = new Set(rows.slice(0, first).map(senderKey));
  }

  if (seenBefore !== null) {
    const fresh = [];
    for (const m of after) {
      const key = senderKey(m);
      if (!seenBefore.has(key) && !isSuspended(m) && !fresh.includes(key)) {
        fresh.push(key);
      }
    }
    if (fresh.length) {
      return ['rerouted',
        `${blocked.length} x 30033 over ${recipients(blocked)} recipient(s), ` +
        `and then ${fresh.join(', ')} started carrying traffic that had never ` +
        'used it before. Moving suspended traffic to another sender is the ' +
        'response that escalates to account termination.'];
    }
  }

  if (later.length) {
    return ['still-pushing',
      `${blocked.length} x 30033 over ${recipients(blocked)} recipient(s), ` +
      `${later.length} of them after the first. The producer has not been told ` +
      `to stop and every one of those is a send that was refused.${partial}`];
  }

  return ['stopped',
    `${blocked.length} x 30033 over ${recipients(blocked)} recipient(s), and ` +
    'nothing refused since. The sending stopped; the suspension is open until ' +
    `Support clears it.${partial}`];
}

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();
}

async function listMessages(auth, account, since, limit = 20000) {
  let url = `${BASE}/Accounts/${account}/Messages.json`;
  let params = { PageSize: 1000, 'DateSent>=': since };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.messages ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

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 days = Number(process.argv.includes('--days')
    ? process.argv[process.argv.indexOf('--days') + 1] : 14) || 14;
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);

  const messages = await listMessages(auth, account, since);
  if (messages.length === 0) {
    console.log(`no messages sent since ${since}`);
    return;
  }

  const [state, detail] = verdict(messages);
  const line = `${state.padEnd(14)} ${detail}  ${messages.length} message(s) ` +
               `since ${since}`;
  if (state === 'clean') { console.log(line); return; }
  console.warn(line);

  const blocked = messages.filter(isSuspended);
  const services = new Set();
  for (const sender of [...new Set(blocked.map(senderKey))].sort()) {
    const count = blocked.filter((m) => senderKey(m) === sender).length;
    console.warn(`  ${sender}  ${count} x 30033`);
    if (sender.startsWith('MG')) services.add(sender);
  }

  for (const service of [...services].sort()) {
    const campaigns = await listV1(auth,
      `${MSG}/Services/${service}/Compliance/Usa2p`, 'compliance');
    const status = campaigns[0]?.campaign_status ?? null;
    console.warn(`  ${service}  campaign_status=${status ?? 'no campaign'}`);
  }

  console.warn('  repair: none by API. Stop the producer, remediate the traffic ' +
               'named in the suspension notice and reply to Twilio Support with ' +
               'evidence. Check the brand above the campaign before assuming the ' +
               'decision was made at the campaign.');
  if (state === 'rerouted') {
    console.warn('  repair: undo the reroute first. Sending the same traffic ' +
                 'from another sender escalates a campaign suspension to the ' +
                 'account.');
  }
  process.exitCode = 1;
}

// 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 datetime import datetime, timedelta, timezone
from email.utils import format_datetime

from twilio_a2p_campaign_suspension_report import (ordered, recipients,
                                                   sender_key, verdict)

T0 = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc)


def at(seconds, **kw):
    """One Message row, dated the way Twilio dates them."""
    row = {"date_sent": format_datetime(T0 + timedelta(seconds=seconds)),
           "messaging_service_sid": "MG1", "to": "+15550000001"}
    row.update(kw)
    return row


def test_a_window_with_no_30033_is_clean():
    state, _ = verdict([at(0), at(60, error_code=30007)])
    assert state == "clean"


def test_sends_continuing_after_the_onset_are_counted_separately():
    state, detail = verdict([at(0), at(60, error_code=30033),
                             at(120, error_code=30033),
                             at(180, error_code=30033)])
    assert state == "still-pushing"
    assert "2 of them after the first" in detail


def test_traffic_that_stopped_after_the_onset_is_its_own_state():
    state, detail = verdict([at(0), at(60, error_code=30033), at(120)])
    assert state == "stopped"
    assert "open until Support" in detail


def test_a_sender_that_appears_only_after_the_onset_is_a_reroute():
    # The dangerous one. MG2 carried nothing before the suspension and is
    # carrying the same traffic afterwards.
    state, detail = verdict([at(0), at(60, error_code=30033),
                             at(120, messaging_service_sid="MG2")])
    assert state == "rerouted"
    assert "MG2" in detail
    assert "termination" in detail


def test_a_window_opening_on_a_30033_refuses_to_guess_at_reroutes():
    # With nothing before the onset every sender looks new, so the check is
    # skipped and the report says to widen the window instead.
    state, detail = verdict([at(0, error_code=30033),
                             at(60, messaging_service_sid="MG2"),
                             at(120, error_code=30033)])
    assert state == "still-pushing"
    assert "widen --days" in detail


def test_retries_are_counted_as_rows_and_recipients_separately():
    rows = [at(0), at(60, error_code=30033), at(70, error_code=30033),
            at(80, error_code=30033)]
    assert recipients(rows[1:]) == 1
    assert "3 x 30033 over 1 recipient(s)" in verdict(rows)[1]


def test_the_messaging_service_wins_over_the_from_number():
    assert sender_key({"messaging_service_sid": "MG1", "from": "+15550001"}) == "MG1"
    assert sender_key({"from": "+15550001"}) == "+15550001"


def test_an_unparseable_date_keeps_its_row_instead_of_dropping_it():
    rows = ordered([{"date_sent": "not a date", "error_code": 30033}, at(0)])
    assert len(rows) == 2
    assert verdict(rows)[0] != "clean"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { ordered, recipients, senderKey, verdict }
  from './twilio-a2p-campaign-suspension-report.mjs';

const T0 = Date.UTC(2026, 7, 24, 12, 0, 0);

/** One Message row, dated the way Twilio dates them. */
const at = (seconds, extra = {}) => ({
  date_sent: new Date(T0 + seconds * 1000).toUTCString(),
  messaging_service_sid: 'MG1',
  to: '+15550000001',
  ...extra,
});

test('a window with no 30033 is clean', () => {
  assert.equal(verdict([at(0), at(60, { error_code: 30007 })])[0], 'clean');
});

test('sends continuing after the onset are counted separately', () => {
  const [state, detail] = verdict([at(0), at(60, { error_code: 30033 }),
    at(120, { error_code: 30033 }), at(180, { error_code: 30033 })]);
  assert.equal(state, 'still-pushing');
  assert.match(detail, /2 of them after the first/);
});

test('traffic that stopped after the onset is its own state', () => {
  const [state, detail] = verdict([at(0), at(60, { error_code: 30033 }), at(120)]);
  assert.equal(state, 'stopped');
  assert.match(detail, /open until Support/);
});

test('a sender that appears only after the onset is a reroute', () => {
  const [state, detail] = verdict([at(0), at(60, { error_code: 30033 }),
    at(120, { messaging_service_sid: 'MG2' })]);
  assert.equal(state, 'rerouted');
  assert.match(detail, /MG2/);
  assert.match(detail, /termination/);
});

test('a window opening on a 30033 refuses to guess at reroutes', () => {
  const [state, detail] = verdict([at(0, { error_code: 30033 }),
    at(60, { messaging_service_sid: 'MG2' }), at(120, { error_code: 30033 })]);
  assert.equal(state, 'still-pushing');
  assert.match(detail, /widen --days/);
});

test('retries are counted as rows and recipients separately', () => {
  const rows = [at(0), at(60, { error_code: 30033 }), at(70, { error_code: 30033 }),
    at(80, { error_code: 30033 })];
  assert.equal(recipients(rows.slice(1)), 1);
  assert.match(verdict(rows)[1], /3 x 30033 over 1 recipient\(s\)/);
});

test('the messaging service wins over the from number', () => {
  assert.equal(senderKey({ messaging_service_sid: 'MG1', from: '+15550001' }), 'MG1');
  assert.equal(senderKey({ from: '+15550001' }), '+15550001');
});

test('an unparseable date keeps its row instead of dropping it', () => {
  const rows = ordered([{ date_sent: 'not a date', error_code: 30033 }, at(0)]);
  assert.equal(rows.length, 2);
  assert.notEqual(verdict(rows)[0], 'clean');
});