an approved regulatory bundle is counting down to expiry

"""Report regulatory Bundles whose approval is about to expire.

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 bundle starts a review you want a human watching.
"""
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_bundle_expiry_audit")

NUMBERS = "https://numbers.twilio.com/v2"

APPROVED = "twilio-approved"
REJECTED = "twilio-rejected"


def parse_date(value):
    """Parse an ISO 8601 timestamp from the numbers v2 API into aware UTC.

    fromisoformat on Python 3.9 does not accept a trailing Z, and a naive
    datetime compared against an aware one raises rather than returning a wrong
    answer, so both are normalised here instead of 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 verdict(bundle, now, horizon_days=60):
    """Classify one Bundle on valid_until. Pure, so the dates can be tested
    without a network and without waiting eighteen months.

    valid_until is the field this whole check exists for: an approved bundle is
    approved as of a date, and when that date passes the bundle is rejected with
    nobody acting. A null valid_until is a regulation that needs no
    re-attestation, which is a healthy state and not a finding.

    Returns (state, detail).
    """
    status = str(bundle.get("status") or "").strip().lower()
    valid_until = parse_date(bundle.get("valid_until"))
    days = None if valid_until is None else (valid_until - now).days

    if status == REJECTED and days is not None and days < 0:
        return ("rejected",
                "valid_until passed %d day(s) ago and the bundle is now %s: this "
                "is the failure after the fact, and the numbers on this bundle "
                "are non-compliant today." % (-days, REJECTED))

    if status != APPROVED:
        return ("not-approved",
                "status is %s, so there is no approval to expire. That is a "
                "different problem from this one." % (status or "unset"))

    if valid_until is None:
        return ("no-expiry",
                "approved with no valid_until: this regulation does not require "
                "periodic re-attestation, so there is no date to watch.")

    if days < 0:
        return ("expired",
                "valid_until passed %d day(s) ago while the status still reads "
                "%s: the flip is not instantaneous, and the numbers on this "
                "bundle are already out of time." % (-days, APPROVED))

    if days <= horizon_days:
        return ("expiring",
                "valid_until is %d day(s) away. Renewal means new supporting "
                "documents, a reassignment and a review, so start now rather "
                "than on the date." % days)

    return ("current", "valid_until is %d day(s) away." % days)


def callback_note(bundle):
    """The reason this arrives as an outage rather than a notification, or None.

    A status_callback does not give warning, because it fires at the transition.
    It is still the difference between finding out at the moment the bundle
    changes and finding out from a customer.
    """
    if str(bundle.get("status_callback") or "").strip():
        return None
    return ("status_callback is unset: when this bundle changes state nothing is "
            "told, so the first signal will be numbers that stopped working.")


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_bundles(session, only_dated=True, limit=500):
    """Page the Bundles list.

    The numbers v2 API returns rows under `results` and an absolute next page in
    `meta.next_page_url`, unlike the 2010-04-01 API's `next_page_uri` path.
    Sorted ascending on valid-until so the first page is the urgent one.
    """
    url = "%s/RegulatoryCompliance/Bundles" % NUMBERS
    params = {"SortBy": "valid-until", "SortDirection": "ASC", "PageSize": 50}
    if only_dated:
        params["HasValidUntilDate"] = "true"
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("results", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--horizon-days", type=int, default=60,
                    help="how far ahead an expiry counts as a finding")
    ap.add_argument("--all", action="store_true",
                    help="include bundles with no valid_until date")
    ap.add_argument("--as-of", default=None,
                    help="ISO date to measure against, for a reproducible run")
    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

    now = parse_date(args.as_of) if args.as_of else None
    if now is None:
        now = datetime.datetime.now(datetime.timezone.utc)

    session = requests.Session()
    session.auth = (key, secret)

    bundles = list_bundles(session, only_dated=not args.all)
    if not bundles:
        log.info("no regulatory bundles with a valid_until on this account")
        return 0

    expired = soon = 0
    for bundle in bundles:
        state, detail = verdict(bundle, now, args.horizon_days)
        label = "%s/%s" % (bundle.get("iso_country") or "??",
                           bundle.get("number_type") or "?")
        line = "%-12s %s  %s  %s" % (state, bundle.get("sid", "?"), label, detail)
        if state in ("current", "no-expiry", "not-approved"):
            log.info(line)
            continue
        if state in ("expired", "rejected"):
            expired += 1
        else:
            soon += 1
        log.warning(line)
        note = callback_note(bundle)
        if note:
            log.warning("  %s", note)
        log.warning("  repair: POST %s/RegulatoryCompliance/SupportingDocuments with "
                    "current paperwork, assign it via POST %s/RegulatoryCompliance/"
                    "Bundles/%s/ItemAssignments, then POST the bundle with "
                    "Status=pending-review", NUMBERS, NUMBERS, bundle.get("sid", "?"))

    log.info("%d bundle(s), %d expired, %d inside the %d day horizon",
             len(bundles), expired, soon, args.horizon_days)
    return 1 if (expired or soon) else 0


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

/**
 * Report regulatory Bundles whose approval is about to expire.
 *
 * 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 bundle starts a review you want a human
 * watching.
 */
const NUMBERS = 'https://numbers.twilio.com/v2';

const APPROVED = 'twilio-approved';
const REJECTED = 'twilio-rejected';

/** Parse an ISO 8601 timestamp from the numbers v2 API. */
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);
}

/**
 * Classify one Bundle on valid_until. Pure, so the dates can be tested without a
 * network and without waiting eighteen months.
 *
 * An approved bundle is approved as of a date, and when that date passes the
 * bundle is rejected with nobody acting. A null valid_until is a regulation that
 * needs no re-attestation: a healthy state, not a finding.
 *
 * Returns [state, detail].
 */
export function verdict(bundle, now, horizonDays = 60) {
  const status = String(bundle.status ?? '').trim().toLowerCase();
  const validUntil = parseDate(bundle.valid_until);
  const days = validUntil === null
    ? null
    : Math.floor((validUntil.getTime() - now.getTime()) / 86400000);

  if (status === REJECTED && days !== null && days < 0) {
    return ['rejected',
      `valid_until passed ${-days} day(s) ago and the bundle is now ${REJECTED}: ` +
      'this is the failure after the fact, and the numbers on this bundle are ' +
      'non-compliant today.'];
  }

  if (status !== APPROVED) {
    return ['not-approved',
      `status is ${status || 'unset'}, so there is no approval to expire. That is ` +
      'a different problem from this one.'];
  }

  if (validUntil === null) {
    return ['no-expiry',
      'approved with no valid_until: this regulation does not require periodic ' +
      're-attestation, so there is no date to watch.'];
  }

  if (days < 0) {
    return ['expired',
      `valid_until passed ${-days} day(s) ago while the status still reads ` +
      `${APPROVED}: the flip is not instantaneous, and the numbers on this bundle ` +
      'are already out of time.'];
  }

  if (days <= horizonDays) {
    return ['expiring',
      `valid_until is ${days} day(s) away. Renewal means new supporting documents, ` +
      'a reassignment and a review, so start now rather than on the date.'];
  }

  return ['current', `valid_until is ${days} day(s) away.`];
}

/** The reason this arrives as an outage rather than a notification, or null. */
export function callbackNote(bundle) {
  if (String(bundle.status_callback ?? '').trim()) return null;
  return 'status_callback is unset: when this bundle changes state nothing is ' +
         'told, so the first signal will be numbers that stopped working.';
}

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

/**
 * The numbers v2 API returns rows under `results` and an absolute next page in
 * `meta.next_page_url`, unlike the 2010-04-01 API's `next_page_uri` path.
 */
export async function listBundles(auth, onlyDated = true, limit = 500) {
  let url = `${NUMBERS}/RegulatoryCompliance/Bundles`;
  let params = { SortBy: 'valid-until', SortDirection: 'ASC', PageSize: 50 };
  if (onlyDated) params.HasValidUntilDate = 'true';
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.results ?? []));
    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 flag = process.argv.indexOf('--horizon-days');
  const horizonDays = flag === -1 ? 60 : Number.parseInt(process.argv[flag + 1], 10);
  const now = new Date();

  const auth = authHeader(key, secret);
  const bundles = await listBundles(auth, !process.argv.includes('--all'));
  if (bundles.length === 0) {
    console.log('no regulatory bundles with a valid_until on this account');
    return;
  }

  let expired = 0;
  let soon = 0;
  for (const bundle of bundles) {
    const [state, detail] = verdict(bundle, now, horizonDays);
    const label = `${bundle.iso_country ?? '??'}/${bundle.number_type ?? '?'}`;
    const line = `${state.padEnd(12)} ${bundle.sid ?? '?'}  ${label}  ${detail}`;
    if (state === 'current' || state === 'no-expiry' || state === 'not-approved') {
      console.log(line);
      continue;
    }
    if (state === 'expired' || state === 'rejected') expired += 1; else soon += 1;
    console.warn(line);
    const note = callbackNote(bundle);
    if (note) console.warn(`  ${note}`);
    console.warn(`  repair: POST ${NUMBERS}/RegulatoryCompliance/SupportingDocuments ` +
                 `with current paperwork, assign it via POST ${NUMBERS}/` +
                 `RegulatoryCompliance/Bundles/${bundle.sid ?? '?'}/ItemAssignments, ` +
                 'then POST the bundle with Status=pending-review');
  }

  console.log(`${bundles.length} bundle(s), ${expired} expired, ${soon} inside the ` +
              `${horizonDays} day horizon`);
  process.exitCode = (expired || soon) ? 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_bundle_expiry_audit import callback_note, parse_date, verdict

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


def make(**kw):
    bundle = {"sid": "BU00000000000000000000000000000001",
              "friendly_name": "DE local address",
              "iso_country": "DE",
              "number_type": "local",
              "status": "twilio-approved",
              "valid_until": "2027-06-01T00:00:00Z",
              "status_callback": "https://ops.example.com/bundle"}
    bundle.update(kw)
    return bundle


def test_iso_dates_from_the_numbers_v2_api_parse_with_a_trailing_z():
    assert parse_date("2027-06-01T00:00:00Z") == \
        datetime.datetime(2027, 6, 1, tzinfo=datetime.timezone.utc)


def test_an_approved_bundle_far_from_its_date_is_current():
    state, detail = verdict(make(), NOW, horizon_days=60)
    assert state == "current"
    assert "275 day(s)" in detail


def test_an_approved_bundle_inside_the_horizon_is_the_warning():
    state, detail = verdict(make(valid_until="2026-09-15T00:00:00Z"), NOW,
                            horizon_days=60)
    assert state == "expiring"
    assert "16 day(s)" in detail


def test_the_horizon_is_the_thing_that_decides():
    bundle = make(valid_until="2026-10-20T00:00:00Z")
    assert verdict(bundle, NOW, horizon_days=30)[0] == "current"
    assert verdict(bundle, NOW, horizon_days=60)[0] == "expiring"


def test_a_date_already_past_is_an_incident_not_a_warning():
    state, detail = verdict(make(valid_until="2026-08-01T00:00:00Z"), NOW)
    assert state == "expired"
    assert "29 day(s) ago" in detail


def test_the_aftermath_reads_as_rejected_rather_than_as_expired():
    state, detail = verdict(make(status="twilio-rejected",
                                 valid_until="2026-07-01T00:00:00Z"), NOW)
    assert state == "rejected"
    assert "non-compliant today" in detail


def test_a_null_valid_until_is_healthy_and_must_not_be_read_as_expired():
    state, detail = verdict(make(valid_until=None), NOW)
    assert state == "no-expiry"
    assert "re-attestation" in detail


def test_a_bundle_that_was_never_approved_is_somebody_else_s_note():
    state, _ = verdict(make(status="pending-review", valid_until=None), NOW)
    assert state == "not-approved"


def test_a_missing_status_callback_is_reported_alongside_the_date():
    assert callback_note(make(status_callback="")) is not None
    assert callback_note(make()) is None

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { callbackNote, parseDate, verdict } from './twilio-bundle-expiry-audit.mjs';

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

const make = (over = {}) => ({
  sid: 'BU00000000000000000000000000000001',
  friendly_name: 'DE local address',
  iso_country: 'DE',
  number_type: 'local',
  status: 'twilio-approved',
  valid_until: '2027-06-01T00:00:00Z',
  status_callback: 'https://ops.example.com/bundle',
  ...over,
});

test('iso dates from the numbers v2 api parse with a trailing z', () => {
  assert.equal(parseDate('2027-06-01T00:00:00Z').toISOString(),
               '2027-06-01T00:00:00.000Z');
});

test('an approved bundle far from its date is current', () => {
  const [state, detail] = verdict(make(), NOW, 60);
  assert.equal(state, 'current');
  assert.match(detail, /275 day\(s\)/);
});

test('an approved bundle inside the horizon is the warning', () => {
  const [state, detail] = verdict(make({ valid_until: '2026-09-15T00:00:00Z' }), NOW, 60);
  assert.equal(state, 'expiring');
  assert.match(detail, /16 day\(s\)/);
});

test('the horizon is the thing that decides', () => {
  const bundle = make({ valid_until: '2026-10-20T00:00:00Z' });
  assert.equal(verdict(bundle, NOW, 30)[0], 'current');
  assert.equal(verdict(bundle, NOW, 60)[0], 'expiring');
});

test('a date already past is an incident not a warning', () => {
  const [state, detail] = verdict(make({ valid_until: '2026-08-01T00:00:00Z' }), NOW);
  assert.equal(state, 'expired');
  assert.match(detail, /29 day\(s\) ago/);
});

test('the aftermath reads as rejected rather than as expired', () => {
  const [state, detail] = verdict(make({
    status: 'twilio-rejected', valid_until: '2026-07-01T00:00:00Z' }), NOW);
  assert.equal(state, 'rejected');
  assert.match(detail, /non-compliant today/);
});

test('a null valid_until is healthy and must not be read as expired', () => {
  const [state, detail] = verdict(make({ valid_until: null }), NOW);
  assert.equal(state, 'no-expiry');
  assert.match(detail, /re-attestation/);
});

test('a bundle that was never approved is somebody else\'s note', () => {
  assert.equal(
    verdict(make({ status: 'pending-review', valid_until: null }), NOW)[0],
    'not-approved');
});

test('a missing status_callback is reported alongside the date', () => {
  assert.notEqual(callbackNote(make({ status_callback: '' })), null);
  assert.equal(callbackNote(make()), null);
});