no Usage Trigger, so overspend runs with nothing watching

"""Report a Twilio account with no Usage Trigger that can actually fire.

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

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

SPEND_CATEGORY = "totalprice"
RECURRING = ("daily", "monthly", "yearly")


def fires_again(trigger):
    """True when the trigger resets itself rather than firing once and stopping.

    An empty recurring is the difference between an alarm and a fuse, and the
    fired fuse looks identical to a working alarm in the console.
    """
    return str(trigger.get("recurring") or "").strip().lower() in RECURRING


def has_callback(trigger):
    """True when crossing the threshold results in a request to something."""
    return bool(str(trigger.get("callback_url") or "").strip())


def by_price(trigger):
    """True when the threshold is money rather than a count of messages or calls."""
    return str(trigger.get("trigger_by") or "").strip().lower() == "price"


def verdict(triggers):
    """Classify an account's Usage Triggers as a set. Pure, so the coverage rules
    can be tested without a network.

    The finding is about the set rather than about any one trigger: an account
    with six triggers and no recurring price cap is as unalarmed as an account
    with none, and the report has to say so.

    Returns (state, detail).
    """
    triggers = list(triggers or [])
    if not triggers:
        return ("none",
                "no usage triggers on this account: nothing on Twilio's side is "
                "watching spend or volume, and nothing will be until somebody "
                "creates one.")

    live = [t for t in triggers if fires_again(t) and has_callback(t)]
    if not live:
        if any(fires_again(t) for t in triggers):
            return ("no-callback",
                    "%d recurring trigger(s), none with a callback_url: the "
                    "threshold is evaluated and no request is ever made, so "
                    "nothing reaches whoever is on call."
                    % len([t for t in triggers if fires_again(t)]))
        fired = [t for t in triggers if str(t.get("date_fired") or "").strip()]
        if fired:
            return ("spent",
                    "%d of %d trigger(s) have fired and none of them recur: the "
                    "fuse blew and was never replaced, and the account has been "
                    "unalarmed ever since." % (len(fired), len(triggers)))
        return ("one-shot",
                "%d trigger(s), none recurring: each fires exactly once and then "
                "sits in the API looking configured." % len(triggers))

    spend = [t for t in live
             if str(t.get("usage_category") or "").strip().lower() == SPEND_CATEGORY
             and by_price(t)]
    if spend:
        return ("covered",
                "%d recurring price trigger(s) on %s with a callback."
                % (len(spend), SPEND_CATEGORY))

    priced = [t for t in live if by_price(t)]
    if priced:
        cats = sorted({str(t.get("usage_category") or "?").strip().lower()
                       for t in priced})
        return ("category-only",
                "price triggers on %s but none on %s: money that leaves through "
                "any other category is unalarmed."
                % (", ".join(cats), SPEND_CATEGORY))

    return ("count-only",
            "%d live trigger(s), all measuring counts rather than price: the same "
            "segment count to a premium destination costs many times more, which "
            "is the whole point of a pumping attack." % len(live))


def suggested_cap(records, multiplier=3.0, floor=5.0):
    """A daily price cap taken from the busiest of the recent days.

    Pure, and separate from the fetch, because the number this prints ends up in
    somebody's repair command and the arithmetic behind it should be readable.
    """
    prices = []
    for record in records:
        try:
            prices.append(float(record.get("price") or 0.0))
        except (TypeError, ValueError):
            continue
    peak = max(prices) if prices else 0.0
    return round(max(floor, peak * multiplier), 2)


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_triggers(session, account, limit=200):
    """Page Usage/Triggers. next_page_uri is a path, not an absolute URL."""
    url = "%s/Accounts/%s/Usage/Triggers.json" % (BASE, account)
    params = {"PageSize": 50}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("usage_triggers", []))
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def daily_spend(session, account, days):
    """One page of daily totalprice records, enough to find the busiest day."""
    start = (datetime.date.today() - datetime.timedelta(days=days)).isoformat()
    page = get(session, "%s/Accounts/%s/Usage/Records/Daily.json" % (BASE, account),
               Category=SPEND_CATEGORY, StartDate=start, PageSize=100)
    return page.get("usage_records", [])


def describe(trigger):
    return "%s %s %s %s recurring=%s callback=%s" % (
        trigger.get("sid", "?"),
        trigger.get("usage_category", "?"),
        trigger.get("trigger_by", "?"),
        trigger.get("trigger_value", "?"),
        trigger.get("recurring") or "none",
        "yes" if has_callback(trigger) else "no",
    )


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--suggest-cap", action="store_true",
                    help="read daily usage records and print a cap based on them")
    ap.add_argument("--days", type=int, default=30,
                    help="how many days of usage the suggested cap looks at")
    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)

    triggers = list_triggers(session, account)
    for trigger in triggers:
        log.info("  %s", describe(trigger))

    state, detail = verdict(triggers)
    if state == "covered":
        log.info("%-14s %s", state, detail)
        return 0

    log.warning("%-14s %s", state, detail)

    cap = "{daily cap}"
    if args.suggest_cap:
        cap = suggested_cap(daily_spend(session, account, args.days))
        log.warning("  busiest recent day times three: %s", cap)

    log.warning("  repair: POST %s/Accounts/%s/Usage/Triggers.json "
                "UsageCategory=totalprice TriggerBy=price TriggerValue=%s "
                "Recurring=daily CallbackUrl=https://your-app.example.com/usage "
                "CallbackMethod=POST", BASE, account, cap)
    log.warning("  then run this against every subaccount: triggers do not inherit")
    return 1


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

/**
 * Report a Twilio account with no Usage Trigger that can actually fire.
 *
 * 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 SPEND_CATEGORY = 'totalprice';
const RECURRING = ['daily', 'monthly', 'yearly'];

/** True when the trigger resets itself rather than firing once and stopping. */
export function firesAgain(trigger) {
  return RECURRING.includes(String(trigger.recurring ?? '').trim().toLowerCase());
}

/** True when crossing the threshold results in a request to something. */
export function hasCallback(trigger) {
  return Boolean(String(trigger.callback_url ?? '').trim());
}

/** True when the threshold is money rather than a count of messages or calls. */
export function byPrice(trigger) {
  return String(trigger.trigger_by ?? '').trim().toLowerCase() === 'price';
}

/**
 * Classify an account's Usage Triggers as a set. Pure, so the coverage rules can
 * be tested without a network. Returns [state, detail].
 */
export function verdict(triggers) {
  const all = [...(triggers ?? [])];
  if (all.length === 0) {
    return ['none',
      "no usage triggers on this account: nothing on Twilio's side is watching " +
      'spend or volume, and nothing will be until somebody creates one.'];
  }

  const live = all.filter((t) => firesAgain(t) && hasCallback(t));
  if (live.length === 0) {
    const recurring = all.filter(firesAgain);
    if (recurring.length) {
      return ['no-callback',
        `${recurring.length} recurring trigger(s), none with a callback_url: the ` +
        'threshold is evaluated and no request is ever made, so nothing reaches ' +
        'whoever is on call.'];
    }
    const fired = all.filter((t) => String(t.date_fired ?? '').trim());
    if (fired.length) {
      return ['spent',
        `${fired.length} of ${all.length} trigger(s) have fired and none of them ` +
        'recur: the fuse blew and was never replaced, and the account has been ' +
        'unalarmed ever since.'];
    }
    return ['one-shot',
      `${all.length} trigger(s), none recurring: each fires exactly once and then ` +
      'sits in the API looking configured.'];
  }

  const spend = live.filter(
    (t) => String(t.usage_category ?? '').trim().toLowerCase() === SPEND_CATEGORY
      && byPrice(t));
  if (spend.length) {
    return ['covered',
      `${spend.length} recurring price trigger(s) on ${SPEND_CATEGORY} with a callback.`];
  }

  const priced = live.filter(byPrice);
  if (priced.length) {
    const cats = [...new Set(priced.map(
      (t) => String(t.usage_category ?? '?').trim().toLowerCase()))].sort();
    return ['category-only',
      `price triggers on ${cats.join(', ')} but none on ${SPEND_CATEGORY}: money ` +
      'that leaves through any other category is unalarmed.'];
  }

  return ['count-only',
    `${live.length} live trigger(s), all measuring counts rather than price: the ` +
    'same segment count to a premium destination costs many times more, which is ' +
    'the whole point of a pumping attack.'];
}

/** A daily price cap taken from the busiest of the recent days. Pure. */
export function suggestedCap(records, multiplier = 3.0, floor = 5.0) {
  const prices = (records ?? [])
    .map((r) => Number.parseFloat(r.price ?? '0'))
    .filter((n) => Number.isFinite(n));
  const peak = prices.length ? Math.max(...prices) : 0;
  return Math.round(Math.max(floor, peak * multiplier) * 100) / 100;
}

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 listTriggers(auth, account, limit = 200) {
  let url = `${BASE}/Accounts/${account}/Usage/Triggers.json`;
  let params = { PageSize: 50 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.usage_triggers ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

function describe(t) {
  return `${t.sid ?? '?'} ${t.usage_category ?? '?'} ${t.trigger_by ?? '?'} ` +
         `${t.trigger_value ?? '?'} recurring=${t.recurring || 'none'} ` +
         `callback=${hasCallback(t) ? 'yes' : 'no'}`;
}

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 suggest = process.argv.includes('--suggest-cap');

  const triggers = await listTriggers(auth, account);
  for (const t of triggers) console.log(`  ${describe(t)}`);

  const [state, detail] = verdict(triggers);
  if (state === 'covered') {
    console.log(`${state.padEnd(14)} ${detail}`);
    return;
  }
  console.warn(`${state.padEnd(14)} ${detail}`);

  let cap = '{daily cap}';
  if (suggest) {
    const since = new Date(Date.now() - 30 * 86400000).toISOString().slice(0, 10);
    const page = await get(auth, `${BASE}/Accounts/${account}/Usage/Records/Daily.json`,
                           { Category: SPEND_CATEGORY, StartDate: since, PageSize: 100 });
    cap = suggestedCap(page.usage_records ?? []);
    console.warn(`  busiest recent day times three: ${cap}`);
  }

  console.warn(`  repair: POST ${BASE}/Accounts/${account}/Usage/Triggers.json ` +
               `UsageCategory=totalprice TriggerBy=price TriggerValue=${cap} ` +
               'Recurring=daily CallbackUrl=https://your-app.example.com/usage ' +
               'CallbackMethod=POST');
  console.warn('  then run this against every subaccount: triggers do not inherit');
  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 twilio_usage_trigger_audit import suggested_cap, verdict


def make(**kw):
    trigger = {"sid": "UT01", "usage_category": "totalprice", "trigger_by": "price",
               "trigger_value": "250", "recurring": "daily",
               "callback_url": "https://ops.example.com/twilio-usage"}
    trigger.update(kw)
    return trigger


def test_an_account_with_no_triggers_is_the_worst_answer():
    state, detail = verdict([])
    assert state == "none"
    assert "nothing" in detail


def test_a_recurring_price_trigger_with_a_callback_is_coverage():
    state, _ = verdict([make()])
    assert state == "covered"


def test_a_one_shot_trigger_that_already_fired_is_a_spent_fuse():
    state, detail = verdict([make(recurring=None,
                                  date_fired="Tue, 18 Apr 2023 09:12:00 +0000")])
    assert state == "spent"
    assert "fuse" in detail


def test_a_one_shot_that_has_not_fired_yet_is_still_not_an_alarm():
    state, _ = verdict([make(recurring="")])
    assert state == "one-shot"


def test_a_recurring_trigger_with_no_callback_url_reaches_nobody():
    state, detail = verdict([make(callback_url="")])
    assert state == "no-callback"
    assert "on call" in detail


def test_price_triggers_on_a_category_but_not_on_totalprice():
    state, detail = verdict([make(usage_category="sms")])
    assert state == "category-only"
    assert "sms" in detail


def test_counting_messages_is_not_capping_money():
    state, detail = verdict([make(trigger_by="count")])
    assert state == "count-only"
    assert "premium" in detail


def test_one_live_price_trigger_outweighs_the_dead_ones_around_it():
    state, _ = verdict([make(recurring=None), make(callback_url=""), make()])
    assert state == "covered"


def test_suggested_cap_is_the_busiest_day_times_three():
    records = [{"price": "12.50"}, {"price": "40.00"}, {"price": "3.10"}]
    assert suggested_cap(records) == 120.0


def test_suggested_cap_falls_back_to_the_floor_on_a_quiet_account():
    assert suggested_cap([]) == 5.0
    assert suggested_cap([{"price": None}, {"price": "not a number"}]) == 5.0

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { suggestedCap, verdict } from './twilio-usage-trigger-audit.mjs';

const make = (over = {}) => ({
  sid: 'UT01',
  usage_category: 'totalprice',
  trigger_by: 'price',
  trigger_value: '250',
  recurring: 'daily',
  callback_url: 'https://ops.example.com/twilio-usage',
  ...over,
});

test('an account with no triggers is the worst answer', () => {
  const [state, detail] = verdict([]);
  assert.equal(state, 'none');
  assert.match(detail, /nothing/);
});

test('a recurring price trigger with a callback is coverage', () => {
  assert.equal(verdict([make()])[0], 'covered');
});

test('a one-shot trigger that already fired is a spent fuse', () => {
  const [state, detail] = verdict([
    make({ recurring: null, date_fired: 'Tue, 18 Apr 2023 09:12:00 +0000' })]);
  assert.equal(state, 'spent');
  assert.match(detail, /fuse/);
});

test('a one-shot that has not fired yet is still not an alarm', () => {
  assert.equal(verdict([make({ recurring: '' })])[0], 'one-shot');
});

test('a recurring trigger with no callback_url reaches nobody', () => {
  const [state, detail] = verdict([make({ callback_url: '' })]);
  assert.equal(state, 'no-callback');
  assert.match(detail, /on call/);
});

test('price triggers on a category but not on totalprice', () => {
  const [state, detail] = verdict([make({ usage_category: 'sms' })]);
  assert.equal(state, 'category-only');
  assert.match(detail, /sms/);
});

test('counting messages is not capping money', () => {
  const [state, detail] = verdict([make({ trigger_by: 'count' })]);
  assert.equal(state, 'count-only');
  assert.match(detail, /premium/);
});

test('one live price trigger outweighs the dead ones around it', () => {
  assert.equal(
    verdict([make({ recurring: null }), make({ callback_url: '' }), make()])[0],
    'covered');
});

test('suggested cap is the busiest day times three', () => {
  assert.equal(suggestedCap([{ price: '12.50' }, { price: '40.00' }, { price: '3.10' }]),
               120.0);
});

test('suggested cap falls back to the floor on a quiet account', () => {
  assert.equal(suggestedCap([]), 5.0);
  assert.equal(suggestedCap([{ price: null }, { price: 'not a number' }]), 5.0);
});