brand failed 30799: the EIN does not match the legal name

"""Report A2P brands rejected because the tax ID and legal name disagree.

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

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

MISMATCH = "30799"

# What the registry actually compares when it resolves a tax identifier against
# public records. Used only when errors[] names no fields, so the report still
# says where to look instead of saying "failed".
IDENTITY_TRIPLE = ("legal company name", "registered business address",
                   "business_registration_identifier")

# An APPROVED brand can still be carrying an identity that was never checked
# against anything external, which is a weaker position than it looks.
WEAK_IDENTITY = ("SELF_DECLARED", "UNVERIFIED")


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 edit_targets(errors):
    """What to correct on the Customer Profile, from the 30799 entries. Pure.

    Prefers what the API named in `fields`, because that is one thing to check
    rather than three. Falls back to the identity triple only when the entry
    named nothing, so the report is never reduced to the word "failed".
    """
    named = []
    saw_mismatch = False
    for err in errors:
        if error_code(err) != MISMATCH:
            continue
        saw_mismatch = True
        for f in (err.get("fields") or []):
            text = str(f).strip()
            if text and text not in named:
                named.append(text)
    if named:
        return named
    return list(IDENTITY_TRIPLE) if saw_mismatch else []


def verdict(brand):
    """Classify one BrandRegistration by what it says about identity. Pure.

    Returns (state, detail).
    """
    status = str(brand.get("status") or "").upper()
    errors = brand.get("errors") or []
    codes = [error_code(e) for e in errors]
    identity = str(brand.get("identity_status") or "").upper()

    if MISMATCH in codes:
        targets = ", ".join(edit_targets(errors))
        return ("identity-mismatch",
                "%s: the registry could not match the submitted identity against "
                "public records. Correct %s on the Customer Profile, not on the "
                "brand." % (MISMATCH, targets))

    if status == "FAILED":
        other = ", ".join(c for c in codes if c) or "no code"
        return ("failed-elsewhere",
                "FAILED on %s, which is not an identity mismatch. The Customer "
                "Profile business details are not the thing to re-check."
                % other)

    if status == "SUSPENDED":
        return ("suspended",
                "brand is SUSPENDED, which is a compliance decision rather than "
                "an identity check. Nothing here is fixed by editing the "
                "profile.")

    if status in ("PENDING", "IN_REVIEW"):
        return ("in-review",
                "brand is %s: the identity lookup has not returned a verdict "
                "yet." % status)

    if status == "APPROVED":
        if identity in WEAK_IDENTITY:
            return ("approved-unverified-identity",
                    "APPROVED with identity_status %s, so the business identity "
                    "was taken as declared rather than matched to a record. A "
                    "later re-vet can still turn up %s." % (identity, MISMATCH))
        return ("approved",
                "APPROVED with identity_status %s" % (identity or "unset"))

    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. Items come back under `data` on this resource."""
    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 = "%-28s %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) or "?", err["url"])
        if state == "identity-mismatch":
            bundle = brand.get("customer_profile_bundle_sid", "BU...")
            log.warning("  read: GET %s/CustomerProfiles/%s/EntityAssignments to "
                        "find the business End-User holding those fields",
                        TRUSTHUB, bundle)
            log.warning("  repair: edit that End-User in Trust Hub so the legal "
                        "name, address and registration identifier match the "
                        "IRS or CRA record exactly, then resubmit brand %s. "
                        "Three resubmissions are free; a fourth returns 21724",
                        sid)

    log.info("%d brand(s), %d with an identity mismatch", len(brands), bad)
    return 1 if bad else 0


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

/**
 * Report A2P brands rejected because the tax ID and legal name disagree.
 *
 * 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 TRUSTHUB = 'https://trusthub.twilio.com/v1';

const MISMATCH = '30799';

// What the registry compares when it resolves a tax identifier against public
// records. Used only when errors[] names no fields.
const IDENTITY_TRIPLE = ['legal company name', 'registered business address',
                         'business_registration_identifier'];

const WEAK_IDENTITY = ['SELF_DECLARED', 'UNVERIFIED'];

/**
 * 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 '';
}

/**
 * What to correct on the Customer Profile, from the 30799 entries. Pure.
 * Prefers what the API named in `fields`; falls back to the identity triple
 * only when the entry named nothing.
 */
export function editTargets(errors) {
  const named = [];
  let sawMismatch = false;
  for (const err of errors) {
    if (errorCode(err) !== MISMATCH) continue;
    sawMismatch = true;
    for (const f of err.fields ?? []) {
      const text = String(f).trim();
      if (text && !named.includes(text)) named.push(text);
    }
  }
  if (named.length) return named;
  return sawMismatch ? [...IDENTITY_TRIPLE] : [];
}

/**
 * Classify one BrandRegistration by what it says about identity. Pure.
 * Returns [state, detail].
 */
export function verdict(brand) {
  const status = String(brand.status ?? '').toUpperCase();
  const errors = brand.errors ?? [];
  const codes = errors.map(errorCode);
  const identity = String(brand.identity_status ?? '').toUpperCase();

  if (codes.includes(MISMATCH)) {
    const targets = editTargets(errors).join(', ');
    return ['identity-mismatch',
      `${MISMATCH}: the registry could not match the submitted identity against ` +
      `public records. Correct ${targets} on the Customer Profile, not on the ` +
      'brand.'];
  }

  if (status === 'FAILED') {
    const other = codes.filter(Boolean).join(', ') || 'no code';
    return ['failed-elsewhere',
      `FAILED on ${other}, which is not an identity mismatch. The Customer ` +
      'Profile business details are not the thing to re-check.'];
  }

  if (status === 'SUSPENDED') {
    return ['suspended',
      'brand is SUSPENDED, which is a compliance decision rather than an ' +
      'identity check. Nothing here is fixed by editing the profile.'];
  }

  if (status === 'PENDING' || status === 'IN_REVIEW') {
    return ['in-review',
      `brand is ${status}: the identity lookup has not returned a verdict yet.`];
  }

  if (status === 'APPROVED') {
    if (WEAK_IDENTITY.includes(identity)) {
      return ['approved-unverified-identity',
        `APPROVED with identity_status ${identity}, so the business identity was ` +
        `taken as declared rather than matched to a record. A later re-vet can ` +
        `still turn up ${MISMATCH}.`];
    }
    return ['approved', `APPROVED with identity_status ${identity || 'unset'}`];
  }

  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. Items come back under `data` on this resource. */
export async function listBrands(auth, limit = 500) {
  let url = `${MSG}/a2p/BrandRegistrations`;
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, { PageSize: 50 });
    out.push(...(page.data ?? []));
    url = 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(28)} ${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 === 'identity-mismatch') {
      const bundle = brand.customer_profile_bundle_sid ?? 'BU...';
      console.warn(`  read: GET ${TRUSTHUB}/CustomerProfiles/${bundle}/` +
                   'EntityAssignments to find the business End-User holding ' +
                   'those fields');
      console.warn('  repair: edit that End-User in Trust Hub so the legal name, ' +
                   'address and registration identifier match the IRS or CRA ' +
                   `record exactly, then resubmit brand ${sid}. Three ` +
                   'resubmissions are free; a fourth returns 21724');
    }
  }

  console.log(`${brands.length} brand(s), ${bad} with an identity mismatch`);
  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_identity_audit import edit_targets, error_code, verdict

FAILED = {"sid": "BN0123456789", "status": "FAILED"}


def test_30799_is_reported_as_an_identity_mismatch():
    state, detail = verdict(dict(FAILED, errors=[{"code": "30799"}]))
    assert state == "identity-mismatch"
    assert "Customer Profile" in detail


def test_the_brand_resource_spells_the_key_code_not_error_code():
    # The campaign resource says error_code. A classifier that reads only one of
    # them reports every brand rejection as unrecognised.
    assert error_code({"code": 30799}) == "30799"
    assert error_code({"error_code": 30799}) == "30799"


def test_named_fields_win_over_the_identity_triple():
    errors = [{"code": "30799", "fields": ["business_registration_identifier"]}]
    assert edit_targets(errors) == ["business_registration_identifier"]


def test_a_30799_with_no_fields_still_says_where_to_look():
    assert edit_targets([{"code": "30799"}]) == [
        "legal company name", "registered business address",
        "business_registration_identifier"]


def test_other_codes_contribute_no_edit_targets():
    assert edit_targets([{"code": "30898"}]) == []


def test_a_brand_failed_on_another_code_is_not_an_identity_mismatch():
    state, detail = verdict(dict(FAILED, errors=[{"code": "30898"}]))
    assert state == "failed-elsewhere"
    assert "30898" in detail


def test_approved_but_self_declared_is_reported():
    state, detail = verdict({"sid": "BN1", "status": "APPROVED",
                             "identity_status": "SELF_DECLARED"})
    assert state == "approved-unverified-identity"
    assert "30799" in detail


def test_a_vetted_brand_is_clean():
    state, _ = verdict({"sid": "BN1", "status": "APPROVED",
                        "identity_status": "VETTED_VERIFIED"})
    assert state == "approved"


def test_suspension_is_not_an_identity_problem():
    state, detail = verdict({"sid": "BN1", "status": "SUSPENDED"})
    assert state == "suspended"
    assert "compliance decision" in detail

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

const FAILED = { sid: 'BN0123456789', status: 'FAILED' };

test('30799 is reported as an identity mismatch', () => {
  const [state, detail] = verdict({ ...FAILED, errors: [{ code: '30799' }] });
  assert.equal(state, 'identity-mismatch');
  assert.match(detail, /Customer Profile/);
});

test('the brand resource spells the key code, not error_code', () => {
  assert.equal(errorCode({ code: 30799 }), '30799');
  assert.equal(errorCode({ error_code: 30799 }), '30799');
});

test('named fields win over the identity triple', () => {
  assert.deepEqual(
    editTargets([{ code: '30799', fields: ['business_registration_identifier'] }]),
    ['business_registration_identifier']);
});

test('a 30799 with no fields still says where to look', () => {
  assert.deepEqual(editTargets([{ code: '30799' }]),
    ['legal company name', 'registered business address',
     'business_registration_identifier']);
});

test('other codes contribute no edit targets', () => {
  assert.deepEqual(editTargets([{ code: '30898' }]), []);
});

test('a brand failed on another code is not an identity mismatch', () => {
  const [state, detail] = verdict({ ...FAILED, errors: [{ code: '30898' }] });
  assert.equal(state, 'failed-elsewhere');
  assert.match(detail, /30898/);
});

test('approved but self declared is reported', () => {
  const [state, detail] = verdict({ sid: 'BN1', status: 'APPROVED',
                                    identity_status: 'SELF_DECLARED' });
  assert.equal(state, 'approved-unverified-identity');
  assert.match(detail, /30799/);
});

test('a vetted brand is clean', () => {
  assert.equal(verdict({ sid: 'BN1', status: 'APPROVED',
                         identity_status: 'VETTED_VERIFIED' })[0], 'approved');
});

test('suspension is not an identity problem', () => {
  const [state, detail] = verdict({ sid: 'BN1', status: 'SUSPENDED' });
  assert.equal(state, 'suspended');
  assert.match(detail, /compliance decision/);
});