a Sole Proprietor brand blocked by an OTP nobody answered

"""Find Sole Proprietor A2P brands whose SMS passcode was never answered.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. Re-sending a passcode costs a customer a
text message and restarts a 24 hour clock, so this script prints that repair
and leaves the decision to send it with a person.
"""
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_sole_prop_otp_audit")

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

SOLE = "SOLE_PROPRIETOR"

# identity_status runs SELF_DECLARED, UNVERIFIED, VERIFIED, VETTED_VERIFIED.
# Only the last two mean the registered handset replied to the passcode.
ANSWERED = ("VERIFIED", "VETTED_VERIFIED")


def parse_time(value):
    """Parse a messaging v1 timestamp. Pure.

    These come back as ISO 8601 with a trailing Z, which
    datetime.fromisoformat did not accept before Python 3.11.
    """
    text = str(value or "").strip()
    if not text:
        return None
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        return datetime.datetime.fromisoformat(text)
    except ValueError:
        return None


def age_hours(date_created, now):
    """Age of a brand in hours, or None when the timestamp is unreadable."""
    created = parse_time(date_created)
    if created is None or now is None:
        return None
    return (now - created).total_seconds() / 3600.0


def verdict(brand, age, window_hours=24.0):
    """Classify one brand registration against the passcode reply window.

    `age` is the brand's age in hours, or None. Taking it as an argument keeps
    the clock out of the classifier, so the boundary at the window and both
    sides of it are testable without freezing time. Returns (state, detail).
    """
    if not brand:
        return ("no-brand", "no brand registration to read.")

    brand_type = str(brand.get("brand_type") or "").upper()
    if brand_type != SOLE:
        return ("not-sole-prop",
                "brand_type is %s: identity here is proved by the customer "
                "profile, and no passcode is ever sent."
                % (brand_type or "unset"))

    status = str(brand.get("status") or "").upper()
    identity = str(brand.get("identity_status") or "").upper()

    if status == "FAILED":
        return ("brand-failed",
                "the brand itself is FAILED. A fresh passcode changes nothing "
                "until the registration is refiled, so read the failure first.")

    if identity in ANSWERED:
        return ("verified",
                "identity_status is %s: the handset replied and identity is "
                "settled." % identity)

    if not identity:
        return ("identity-unknown",
                "identity_status is not set on this brand, so nothing can be "
                "concluded about the passcode from this response.")

    links = brand.get("links") or {}
    if not links.get("brand_registration_otps"):
        return ("no-otp-subresource",
                "identity_status is %s and links.brand_registration_otps is "
                "absent, so no passcode has been raised on this brand at all. "
                "This is a submission problem, not an unanswered text."
                % identity)

    if age is None:
        return ("age-unknown",
                "identity_status is %s and date_created could not be read, so "
                "this cannot be aged against the reply window." % identity)

    if age >= window_hours:
        return ("otp-lapsed",
                "identity_status is still %s, %.0f hours after the brand was "
                "created. The %.0f hour reply window has closed and the "
                "passcode expired unanswered. status reads %s, which is not "
                "the field that unblocks sending."
                % (identity, age, window_hours, status or "unset"))

    return ("otp-outstanding",
            "identity_status is %s, %.0f hours in. The owner has about %.0f "
            "hours left to reply from the registered handset."
            % (identity, age, window_hours - age))


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("--window-hours", type=float, default=24.0,
                    help="how long the owner has to reply to the passcode")
    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)
    now = datetime.datetime.now(datetime.timezone.utc)

    brands = list_v1(session, MSG + "/a2p/BrandRegistrations", "data",
                     args.max_brands)
    if not brands:
        log.info("no A2P brand registrations on this account")
        return 0

    sole = 0
    bad = 0
    for brand in brands:
        age = age_hours(brand.get("date_created"), now)
        state, detail = verdict(brand, age, args.window_hours)
        if state == "not-sole-prop":
            continue
        sole += 1
        name = brand.get("brand_sid") or brand.get("sid") or "brand"
        line = "%-20s %s  %s" % (state, name, detail)
        if state in ("verified", "otp-outstanding"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if state in ("otp-lapsed", "age-unknown"):
            log.warning("  repair: raise a fresh passcode at %s/a2p/"
                        "BrandRegistrations/%s/SmsOtp, then have the owner "
                        "reply from the registered handset within %.0f hours",
                        MSG, name, args.window_hours)
            log.warning("  repair: if that mobile already backs three A2P brand "
                        "registrations anywhere in the registry, or is not a "
                        "real US or Canadian handset, refile the profile with a "
                        "different number instead")
        elif state == "no-otp-subresource":
            log.warning("  repair: check how this brand was submitted before "
                        "sending anything; there is no passcode to re-send")

    log.info("%d brand(s), %d sole proprietor, %d waiting on a passcode",
             len(brands), sole, bad)
    return 1 if bad else 0


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

/**
 * Find Sole Proprietor A2P brands whose SMS passcode was never answered.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The passcode re-send is printed,
 * never performed.
 */
const MSG = 'https://messaging.twilio.com/v1';

const SOLE = 'SOLE_PROPRIETOR';

// identity_status runs SELF_DECLARED, UNVERIFIED, VERIFIED, VETTED_VERIFIED.
// Only the last two mean the registered handset replied.
const ANSWERED = ['VERIFIED', 'VETTED_VERIFIED'];

/** Parse a messaging v1 ISO 8601 timestamp. Pure. Returns a Date or null. */
export function parseTime(value) {
  const text = String(value ?? '').trim();
  if (!text) return null;
  const t = Date.parse(text);
  return Number.isNaN(t) ? null : new Date(t);
}

/** Age of a brand in hours, or null when the timestamp is unreadable. */
export function ageHours(dateCreated, now) {
  const created = parseTime(dateCreated);
  if (created === null || !now) return null;
  return (now.getTime() - created.getTime()) / 3600000;
}

/**
 * Classify one brand registration against the passcode reply window. `age` is
 * in hours, or null; taking it as an argument keeps the clock out of the
 * classifier. Pure. Returns [state, detail].
 */
export function verdict(brand, age, windowHours = 24.0) {
  if (!brand) return ['no-brand', 'no brand registration to read.'];

  const brandType = String(brand.brand_type ?? '').toUpperCase();
  if (brandType !== SOLE) {
    return ['not-sole-prop',
      `brand_type is ${brandType || 'unset'}: identity here is proved by the ` +
      'customer profile, and no passcode is ever sent.'];
  }

  const status = String(brand.status ?? '').toUpperCase();
  const identity = String(brand.identity_status ?? '').toUpperCase();

  if (status === 'FAILED') {
    return ['brand-failed',
      'the brand itself is FAILED. A fresh passcode changes nothing until the ' +
      'registration is refiled, so read the failure first.'];
  }

  if (ANSWERED.includes(identity)) {
    return ['verified',
      `identity_status is ${identity}: the handset replied and identity is settled.`];
  }

  if (!identity) {
    return ['identity-unknown',
      'identity_status is not set on this brand, so nothing can be concluded ' +
      'about the passcode from this response.'];
  }

  const links = brand.links ?? {};
  if (!links.brand_registration_otps) {
    return ['no-otp-subresource',
      `identity_status is ${identity} and links.brand_registration_otps is ` +
      'absent, so no passcode has been raised on this brand at all. This is a ' +
      'submission problem, not an unanswered text.'];
  }

  if (age === null) {
    return ['age-unknown',
      `identity_status is ${identity} and date_created could not be read, so ` +
      'this cannot be aged against the reply window.'];
  }

  if (age >= windowHours) {
    return ['otp-lapsed',
      `identity_status is still ${identity}, ${age.toFixed(0)} hours after the ` +
      `brand was created. The ${windowHours.toFixed(0)} hour reply window has ` +
      'closed and the passcode expired unanswered. status reads ' +
      `${status || 'unset'}, which is not the field that unblocks sending.`];
  }

  return ['otp-outstanding',
    `identity_status is ${identity}, ${age.toFixed(0)} hours in. The owner has ` +
    `about ${(windowHours - age).toFixed(0)} hours left to reply from the ` +
    'registered handset.'];
}

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 flag = process.argv.indexOf('--window-hours');
  const windowHours = flag >= 0 ? Number(process.argv[flag + 1]) : 24.0;
  const now = new Date();

  const brands = await listV1(auth, `${MSG}/a2p/BrandRegistrations`, 'data');
  if (brands.length === 0) {
    console.log('no A2P brand registrations on this account');
    return;
  }

  let sole = 0;
  let bad = 0;
  for (const brand of brands) {
    const age = ageHours(brand.date_created, now);
    const [state, detail] = verdict(brand, age, windowHours);
    if (state === 'not-sole-prop') continue;
    sole += 1;
    const name = brand.brand_sid ?? brand.sid ?? 'brand';
    const line = `${state.padEnd(20)} ${name}  ${detail}`;
    if (state === 'verified' || state === 'otp-outstanding') {
      console.log(line);
      continue;
    }
    bad += 1;
    console.warn(line);
    if (state === 'otp-lapsed' || state === 'age-unknown') {
      console.warn(`  repair: raise a fresh passcode at ${MSG}/a2p/` +
                   `BrandRegistrations/${name}/SmsOtp, then have the owner reply ` +
                   `from the registered handset within ${windowHours} hours`);
      console.warn('  repair: if that mobile already backs three A2P brand ' +
                   'registrations anywhere in the registry, or is not a real US ' +
                   'or Canadian handset, refile the profile with a different number');
    } else if (state === 'no-otp-subresource') {
      console.warn('  repair: check how this brand was submitted before sending ' +
                   'anything; there is no passcode to re-send');
    }
  }

  console.log(`${brands.length} brand(s), ${sole} sole proprietor, ${bad} waiting ` +
              'on a passcode');
  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_sole_prop_otp_audit import age_hours, verdict

OTPS = {"brand_registration_otps": "https://messaging.twilio.com/v1/a2p/"
                                   "BrandRegistrations/BN01/SmsOtp"}
WAITING = {"brand_type": "SOLE_PROPRIETOR", "status": "PENDING",
           "identity_status": "SELF_DECLARED", "links": OTPS}
NOW = datetime.datetime(2026, 8, 30, tzinfo=datetime.timezone.utc)


def test_inside_the_window_the_passcode_is_still_in_flight():
    state, detail = verdict(WAITING, 6.0)
    assert state == "otp-outstanding"
    assert "18 hours left" in detail


def test_past_the_window_the_passcode_has_expired_unanswered():
    state, detail = verdict(WAITING, 40.0)
    assert state == "otp-lapsed"
    assert "reply window has closed" in detail


def test_approved_status_does_not_rescue_an_unverified_identity():
    # This is the state that reaches production: somebody reads status, sees
    # APPROVED, and enables US sending on a brand that cannot register numbers.
    state, detail = verdict(dict(WAITING, status="APPROVED"), 72.0)
    assert state == "otp-lapsed"
    assert "status reads APPROVED" in detail


def test_vetted_verified_counts_as_answered():
    assert verdict(dict(WAITING, identity_status="VETTED_VERIFIED"), 500.0)[0] == "verified"


def test_missing_otp_subresource_is_not_an_unanswered_text():
    brand = dict(WAITING, links={})
    state, detail = verdict(brand, 200.0)
    assert state == "no-otp-subresource"
    assert "submission problem" in detail


def test_a_failed_brand_is_read_before_the_passcode():
    assert verdict(dict(WAITING, status="FAILED"), 200.0)[0] == "brand-failed"


def test_standard_brands_are_left_alone():
    state, detail = verdict({"brand_type": "STANDARD", "identity_status": "UNVERIFIED"}, 999.0)
    assert state == "not-sole-prop"
    assert "no passcode is ever sent" in detail


def test_an_unreadable_date_is_reported_rather_than_guessed():
    assert verdict(WAITING, None)[0] == "age-unknown"
    assert age_hours("not a date", NOW) is None
    assert round(age_hours("2026-08-29T00:00:00Z", NOW)) == 24

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { ageHours, verdict } from './twilio-sole-prop-otp-audit.mjs';

const OTPS = {
  brand_registration_otps:
    'https://messaging.twilio.com/v1/a2p/BrandRegistrations/BN01/SmsOtp',
};
const WAITING = {
  brand_type: 'SOLE_PROPRIETOR',
  status: 'PENDING',
  identity_status: 'SELF_DECLARED',
  links: OTPS,
};
const NOW = new Date('2026-08-30T00:00:00Z');

test('inside the window the passcode is still in flight', () => {
  const [state, detail] = verdict(WAITING, 6.0);
  assert.equal(state, 'otp-outstanding');
  assert.match(detail, /18 hours left/);
});

test('past the window the passcode has expired unanswered', () => {
  const [state, detail] = verdict(WAITING, 40.0);
  assert.equal(state, 'otp-lapsed');
  assert.match(detail, /reply window has closed/);
});

test('approved status does not rescue an unverified identity', () => {
  const [state, detail] = verdict({ ...WAITING, status: 'APPROVED' }, 72.0);
  assert.equal(state, 'otp-lapsed');
  assert.match(detail, /status reads APPROVED/);
});

test('vetted verified counts as answered', () => {
  assert.equal(verdict({ ...WAITING, identity_status: 'VETTED_VERIFIED' }, 500.0)[0],
               'verified');
});

test('missing otp subresource is not an unanswered text', () => {
  const [state, detail] = verdict({ ...WAITING, links: {} }, 200.0);
  assert.equal(state, 'no-otp-subresource');
  assert.match(detail, /submission problem/);
});

test('a failed brand is read before the passcode', () => {
  assert.equal(verdict({ ...WAITING, status: 'FAILED' }, 200.0)[0], 'brand-failed');
});

test('standard brands are left alone', () => {
  const [state, detail] = verdict(
    { brand_type: 'STANDARD', identity_status: 'UNVERIFIED' }, 999.0);
  assert.equal(state, 'not-sole-prop');
  assert.match(detail, /no passcode is ever sent/);
});

test('an unreadable date is reported rather than guessed', () => {
  assert.equal(verdict(WAITING, null)[0], 'age-unknown');
  assert.equal(ageHours('not a date', NOW), null);
  assert.equal(Math.round(ageHours('2026-08-29T00:00:00Z', NOW)), 24);
});