one rejected Trust Hub profile fails brands and toll-free

"""Report Trust Hub Customer Profiles that block A2P brands and toll-free.

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 resubmitting a profile starts a review and re-triggering a brand
consumes one of a small number of free attempts.
"""
import argparse
import datetime
import logging
import os
import sys

import requests

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

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

APPROVED = "twilio-approved"
REJECTED = "twilio-rejected"
DRAFT = "draft"
REVIEWING = ("pending-review", "in-review")


def parse_date(value):
    """Parse an ISO 8601 timestamp into aware UTC.

    fromisoformat on Python 3.9 rejects a trailing Z, and comparing naive to
    aware raises, so both are normalised here rather than at every call site.
    """
    text = str(value or "").strip()
    if not text:
        return None
    try:
        parsed = datetime.datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=datetime.timezone.utc)
    return parsed.astimezone(datetime.timezone.utc)


def error_lines(profile):
    """Readable lines from the profile's errors, whatever shape they arrive in.

    Entries are objects with a code and a description, but a bare string turns up
    too, and str() on a dict in a report is worse than useless.
    """
    out = []
    for err in profile.get("errors") or []:
        if isinstance(err, dict):
            code = err.get("code") or err.get("error_code") or "no code"
            text = (err.get("description") or err.get("message")
                    or "no description")
            out.append("%s: %s" % (code, text))
        else:
            out.append(str(err))
    return out


def verdict(profile, now):
    """Classify one Customer Profile. Pure, so the states can be tested offline.

    An approved profile past valid_until is the case worth having its own state:
    the status still reads approved, and everything built on it has stopped
    inheriting an approval that no longer exists.

    Returns (state, detail).
    """
    status = str(profile.get("status") or "").strip().lower()
    valid_until = parse_date(profile.get("valid_until"))

    if status == REJECTED:
        return ("rejected",
                "twilio-rejected: every product built on this profile fails "
                "downstream in its own vocabulary. The reason is in errors on "
                "this object, not on the brand or the verification.")

    if status == DRAFT:
        return ("draft",
                "still a draft: never submitted, so never reviewed and never "
                "rejected. It blocks the same downstream products, and it has no "
                "errors to read because nothing has looked at it.")

    if status in REVIEWING:
        return ("in-review",
                "%s: submitted and waiting. Downstream submissions made now will "
                "fail, so this is a reason to hold them rather than to retry "
                "them." % status)

    if status == APPROVED:
        if valid_until is not None and valid_until <= now:
            return ("expired",
                    "status still reads twilio-approved but valid_until passed "
                    "on %s: the approval that downstream products inherited is "
                    "gone." % valid_until.date().isoformat())
        return ("approved", "twilio-approved and in date.")

    return ("unknown",
            "status is %s, which this script does not classify. Read it rather "
            "than assuming it is healthy." % (status or "unset"))


def dependents(profile_sid, brands, verifications):
    """Name what stops working while this profile is not approved. Pure.

    The two products spell the same reference differently: brands use
    customer_profile_bundle_sid, toll-free verifications use
    customer_profile_sid. A join written for one matches nothing on the other,
    which is how half the blast radius goes unreported.
    """
    sid = str(profile_sid or "").strip()
    if not sid:
        return []
    out = []
    for brand in brands or []:
        if str(brand.get("customer_profile_bundle_sid") or "").strip() == sid:
            out.append("brand %s (%s)" % (brand.get("sid", "?"),
                                          brand.get("status") or "no status"))
    for record in verifications or []:
        if str(record.get("customer_profile_sid") or "").strip() == sid:
            out.append("toll-free verification %s (%s)"
                       % (record.get("sid", "?"),
                          record.get("status") or "no status"))
    return out


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=500):
    """Page a v1 collection. The items key differs per resource: Trust Hub uses
    `results`, BrandRegistrations uses `data`, toll-free uses `verifications`.
    meta.next_page_url is absolute in all three."""
    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 main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--all", action="store_true",
                    help="list approved profiles too, with their dependants")
    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)
    now = datetime.datetime.now(datetime.timezone.utc)

    profiles = list_v1(session, TRUSTHUB + "/CustomerProfiles", "results")
    if not profiles:
        log.info("no Trust Hub customer profiles on this account")
        return 0

    brands = list_v1(session, MSG + "/a2p/BrandRegistrations", "data")
    verifications = list_v1(session, MSG + "/Tollfree/Verifications", "verifications")

    bad = blocked = 0
    for profile in profiles:
        sid = profile.get("sid", "?")
        state, detail = verdict(profile, now)
        downstream = dependents(sid, brands, verifications)
        line = "%-10s %s  %s  %s" % (state, sid,
                                     profile.get("friendly_name") or "no name",
                                     detail)
        if state == "approved" and not args.all:
            log.info(line)
            continue
        if state == "approved":
            log.info(line)
            for item in downstream:
                log.info("  built on this profile: %s", item)
            continue

        bad += 1
        blocked += len(downstream)
        log.warning(line)
        for text in error_lines(profile):
            log.warning("  error %s", text)
        for item in downstream:
            log.warning("  blocked: %s", item)
        if not downstream:
            log.warning("  nothing downstream references this profile yet, which "
                        "makes it a ticket rather than an outage")
        log.warning("  repair: correct the objects at %s/CustomerProfiles/%s/"
                    "EntityAssignments, send the profile back with "
                    "Status=pending-review, and re-trigger the brand or "
                    "verification only once it is approved", TRUSTHUB, sid)

    log.info("%d profile(s), %d blocking %d downstream object(s)",
             len(profiles), bad, blocked)
    return 1 if bad else 0


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

/**
 * Report Trust Hub Customer Profiles that block A2P brands and toll-free.
 *
 * 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 resubmitting a profile starts a review and re-triggering a
 * brand consumes one of a small number of free attempts.
 */
const TRUSTHUB = 'https://trusthub.twilio.com/v1';
const MSG = 'https://messaging.twilio.com/v1';

const APPROVED = 'twilio-approved';
const REJECTED = 'twilio-rejected';
const DRAFT = 'draft';
const REVIEWING = ['pending-review', 'in-review'];

/** Parse an ISO 8601 timestamp. */
export function parseDate(value) {
  const text = String(value ?? '').trim();
  if (!text) return null;
  const ms = Date.parse(text);
  return Number.isNaN(ms) ? null : new Date(ms);
}

/**
 * Readable lines from the profile's errors, whatever shape they arrive in.
 * Entries are objects with a code and a description, but a bare string turns up
 * too, and stringifying an object in a report is worse than useless.
 */
export function errorLines(profile) {
  const out = [];
  for (const err of profile.errors ?? []) {
    if (err && typeof err === 'object') {
      const code = err.code ?? err.error_code ?? 'no code';
      const text = err.description ?? err.message ?? 'no description';
      out.push(`${code}: ${text}`);
    } else {
      out.push(String(err));
    }
  }
  return out;
}

/**
 * Classify one Customer Profile. Pure, so the states can be tested offline.
 *
 * An approved profile past valid_until is the case worth its own state: the
 * status still reads approved, and everything built on it has stopped
 * inheriting an approval that no longer exists.
 *
 * Returns [state, detail].
 */
export function verdict(profile, now) {
  const status = String(profile.status ?? '').trim().toLowerCase();
  const validUntil = parseDate(profile.valid_until);

  if (status === REJECTED) {
    return ['rejected',
      'twilio-rejected: every product built on this profile fails downstream in ' +
      'its own vocabulary. The reason is in errors on this object, not on the ' +
      'brand or the verification.'];
  }

  if (status === DRAFT) {
    return ['draft',
      'still a draft: never submitted, so never reviewed and never rejected. It ' +
      'blocks the same downstream products, and it has no errors to read ' +
      'because nothing has looked at it.'];
  }

  if (REVIEWING.includes(status)) {
    return ['in-review',
      `${status}: submitted and waiting. Downstream submissions made now will ` +
      'fail, so this is a reason to hold them rather than to retry them.'];
  }

  if (status === APPROVED) {
    if (validUntil !== null && validUntil <= now) {
      return ['expired',
        'status still reads twilio-approved but valid_until passed on ' +
        `${validUntil.toISOString().slice(0, 10)}: the approval that downstream ` +
        'products inherited is gone.'];
    }
    return ['approved', 'twilio-approved and in date.'];
  }

  return ['unknown',
    `status is ${status || 'unset'}, which this script does not classify. Read ` +
    'it rather than assuming it is healthy.'];
}

/**
 * Name what stops working while this profile is not approved. Pure.
 *
 * The two products spell the same reference differently: brands use
 * customer_profile_bundle_sid, toll-free verifications use customer_profile_sid.
 * A join written for one matches nothing on the other.
 */
export function dependents(profileSid, brands, verifications) {
  const sid = String(profileSid ?? '').trim();
  if (!sid) return [];
  const out = [];
  for (const brand of brands ?? []) {
    if (String(brand.customer_profile_bundle_sid ?? '').trim() === sid) {
      out.push(`brand ${brand.sid ?? '?'} (${brand.status ?? 'no status'})`);
    }
  }
  for (const record of verifications ?? []) {
    if (String(record.customer_profile_sid ?? '').trim() === sid) {
      out.push(`toll-free verification ${record.sid ?? '?'} ` +
               `(${record.status ?? 'no status'})`);
    }
  }
  return out;
}

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 a v1 collection. The items key differs per resource: Trust Hub uses
 * `results`, BrandRegistrations uses `data`, toll-free uses `verifications`.
 */
export async function listV1(auth, url, key, limit = 500) {
  let params = { PageSize: 50 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page[key] ?? []));
    url = page.meta?.next_page_url ?? 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 all = process.argv.includes('--all');
  const now = new Date();

  const profiles = await listV1(auth, `${TRUSTHUB}/CustomerProfiles`, 'results');
  if (profiles.length === 0) {
    console.log('no Trust Hub customer profiles on this account');
    return;
  }

  const brands = await listV1(auth, `${MSG}/a2p/BrandRegistrations`, 'data');
  const verifications = await listV1(auth, `${MSG}/Tollfree/Verifications`,
                                     'verifications');

  let bad = 0;
  let blocked = 0;
  for (const profile of profiles) {
    const sid = profile.sid ?? '?';
    const [state, detail] = verdict(profile, now);
    const downstream = dependents(sid, brands, verifications);
    const line = `${state.padEnd(10)} ${sid}  ` +
                 `${profile.friendly_name ?? 'no name'}  ${detail}`;
    if (state === 'approved') {
      console.log(line);
      if (all) for (const item of downstream) {
        console.log(`  built on this profile: ${item}`);
      }
      continue;
    }

    bad += 1;
    blocked += downstream.length;
    console.warn(line);
    for (const text of errorLines(profile)) console.warn(`  error ${text}`);
    for (const item of downstream) console.warn(`  blocked: ${item}`);
    if (downstream.length === 0) {
      console.warn('  nothing downstream references this profile yet, which ' +
                   'makes it a ticket rather than an outage');
    }
    console.warn(`  repair: correct the objects at ${TRUSTHUB}/CustomerProfiles/` +
                 `${sid}/EntityAssignments, send the profile back with ` +
                 'Status=pending-review, and re-trigger the brand or verification ' +
                 'only once it is approved');
  }

  console.log(`${profiles.length} profile(s), ${bad} blocking ` +
              `${blocked} downstream object(s)`);
  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; });
}

import datetime

from twilio_customer_profile_audit import (dependents, error_lines, verdict)

NOW = datetime.datetime(2026, 8, 30, tzinfo=datetime.timezone.utc)

PROFILE_SID = "BU00000000000000000000000000000003"

BRANDS = [
    {"sid": "BN00000000000000000000000000000001",
     "status": "FAILED",
     "customer_profile_bundle_sid": PROFILE_SID},
    {"sid": "BN00000000000000000000000000000002",
     "status": "APPROVED",
     "customer_profile_bundle_sid": "BU99999999999999999999999999999999"},
]

VERIFICATIONS = [
    {"sid": "HH00000000000000000000000000000001",
     "status": "TWILIO_REJECTED",
     "customer_profile_sid": PROFILE_SID},
    {"sid": "HH00000000000000000000000000000002",
     "status": "TWILIO_APPROVED",
     "customer_profile_sid": "BU99999999999999999999999999999999"},
]


def make(**kw):
    profile = {"sid": PROFILE_SID,
               "friendly_name": "Example Ltd primary",
               "status": "twilio-rejected",
               "valid_until": None,
               "policy_sid": "RN00000000000000000000000000000001",
               "errors": []}
    profile.update(kw)
    return profile


def test_a_rejected_profile_points_at_itself_rather_than_downstream():
    state, detail = verdict(make(), NOW)
    assert state == "rejected"
    assert "not on the brand" in detail


def test_draft_blocks_the_same_products_and_has_no_errors_to_read():
    state, detail = verdict(make(status="draft"), NOW)
    assert state == "draft"
    assert "never submitted" in detail


def test_review_states_are_a_reason_to_hold_not_to_retry():
    state, detail = verdict(make(status="in-review"), NOW)
    assert state == "in-review"
    assert "hold them" in detail
    assert verdict(make(status="pending-review"), NOW)[0] == "in-review"


def test_an_approved_profile_past_valid_until_is_not_approved():
    state, detail = verdict(
        make(status="twilio-approved", valid_until="2026-07-01T00:00:00Z"), NOW)
    assert state == "expired"
    assert "2026-07-01" in detail


def test_an_approved_profile_in_date_is_the_only_healthy_state():
    state, _ = verdict(
        make(status="twilio-approved", valid_until="2027-07-01T00:00:00Z"), NOW)
    assert state == "approved"
    state, _ = verdict(make(status="twilio-approved", valid_until=None), NOW)
    assert state == "approved"


def test_dependents_match_both_spellings_of_the_same_reference():
    found = dependents(PROFILE_SID, BRANDS, VERIFICATIONS)
    assert len(found) == 2
    assert "brand BN00000000000000000000000000000001 (FAILED)" in found
    assert any(f.startswith("toll-free verification HH00") for f in found)


def test_objects_on_another_profile_are_not_claimed():
    assert dependents("BU00000000000000000000000000000009",
                      BRANDS, VERIFICATIONS) == []
    assert dependents("", BRANDS, VERIFICATIONS) == []
    assert dependents(PROFILE_SID, None, None) == []


def test_errors_render_whether_they_are_objects_or_strings():
    lines = error_lines(make(errors=[{"code": 21212,
                                      "description": "business name mismatch"},
                                     "legacy string entry"]))
    assert lines[0] == "21212: business name mismatch"
    assert lines[1] == "legacy string entry"
    assert error_lines(make()) == []

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { dependents, errorLines, verdict } from './twilio-customer-profile-audit.mjs';

const NOW = new Date('2026-08-30T00:00:00Z');
const PROFILE_SID = 'BU00000000000000000000000000000003';

const BRANDS = [
  { sid: 'BN00000000000000000000000000000001', status: 'FAILED',
    customer_profile_bundle_sid: PROFILE_SID },
  { sid: 'BN00000000000000000000000000000002', status: 'APPROVED',
    customer_profile_bundle_sid: 'BU99999999999999999999999999999999' },
];

const VERIFICATIONS = [
  { sid: 'HH00000000000000000000000000000001', status: 'TWILIO_REJECTED',
    customer_profile_sid: PROFILE_SID },
  { sid: 'HH00000000000000000000000000000002', status: 'TWILIO_APPROVED',
    customer_profile_sid: 'BU99999999999999999999999999999999' },
];

const make = (over = {}) => ({
  sid: PROFILE_SID,
  friendly_name: 'Example Ltd primary',
  status: 'twilio-rejected',
  valid_until: null,
  policy_sid: 'RN00000000000000000000000000000001',
  errors: [],
  ...over,
});

test('a rejected profile points at itself rather than downstream', () => {
  const [state, detail] = verdict(make(), NOW);
  assert.equal(state, 'rejected');
  assert.match(detail, /not on the brand/);
});

test('draft blocks the same products and has no errors to read', () => {
  const [state, detail] = verdict(make({ status: 'draft' }), NOW);
  assert.equal(state, 'draft');
  assert.match(detail, /never submitted/);
});

test('review states are a reason to hold not to retry', () => {
  const [state, detail] = verdict(make({ status: 'in-review' }), NOW);
  assert.equal(state, 'in-review');
  assert.match(detail, /hold them/);
  assert.equal(verdict(make({ status: 'pending-review' }), NOW)[0], 'in-review');
});

test('an approved profile past valid_until is not approved', () => {
  const [state, detail] = verdict(
    make({ status: 'twilio-approved', valid_until: '2026-07-01T00:00:00Z' }), NOW);
  assert.equal(state, 'expired');
  assert.match(detail, /2026-07-01/);
});

test('an approved profile in date is the only healthy state', () => {
  assert.equal(verdict(
    make({ status: 'twilio-approved', valid_until: '2027-07-01T00:00:00Z' }),
    NOW)[0], 'approved');
  assert.equal(verdict(
    make({ status: 'twilio-approved', valid_until: null }), NOW)[0], 'approved');
});

test('dependents match both spellings of the same reference', () => {
  const found = dependents(PROFILE_SID, BRANDS, VERIFICATIONS);
  assert.equal(found.length, 2);
  assert.ok(found.includes('brand BN00000000000000000000000000000001 (FAILED)'));
  assert.ok(found.some((f) => f.startsWith('toll-free verification HH00')));
});

test('objects on another profile are not claimed', () => {
  assert.deepEqual(
    dependents('BU00000000000000000000000000000009', BRANDS, VERIFICATIONS), []);
  assert.deepEqual(dependents('', BRANDS, VERIFICATIONS), []);
  assert.deepEqual(dependents(PROFILE_SID, null, null), []);
});

test('errors render whether they are objects or strings', () => {
  const lines = errorLines(make({
    errors: [{ code: 21212, description: 'business name mismatch' },
             'legacy string entry'] }));
  assert.equal(lines[0], '21212: business name mismatch');
  assert.equal(lines[1], 'legacy string entry');
  assert.deepEqual(errorLines(make()), []);
});