no endpoint subscribes to any payment failure event

"""Report whether anything subscribes to Stripe payment and invoice failure events.

Read only. Three GETs, no writes: give this a RESTRICTED key with read access to
Webhook Endpoints, Subscriptions and Events. The repair is printed, never
performed, because this script holds a credential to a live payments account.
"""
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("stripe_payment_failure_events")

API = "https://api.stripe.com/v1"

PI_OK = "payment_intent.succeeded"
PI_FAIL = "payment_intent.payment_failed"
INV_FAIL = "invoice.payment_failed"


def verdict(subscribed, has_active_subscriptions, failed_invoices):
    """Classify payment-failure coverage across both surfaces. Pure and testable.

    `subscribed` is the union of enabled_events across every endpoint. The
    billing surface only counts when the account actually has recurring billing,
    and failures already seen turn a gap into an incident. Returns (state, detail).
    """
    events = set(subscribed or [])
    if "*" in events:
        return ("wildcard",
                "a wildcard covers both failure events, and every other type "
                "along with them.")

    one_off_gap = PI_OK in events and PI_FAIL not in events
    billing_gap = bool(has_active_subscriptions) and INV_FAIL not in events

    if billing_gap and failed_invoices:
        return ("blind",
                "%d invoice payment(s) already failed and %s is not subscribed. "
                "Dunning is running right now and nothing is being told."
                % (failed_invoices, INV_FAIL))
    if one_off_gap and billing_gap:
        return ("exposed",
                "neither %s nor %s is subscribed. Both the one-off and the "
                "billing failure paths are silent." % (PI_FAIL, INV_FAIL))
    if one_off_gap:
        return ("one-sided",
                "%s is subscribed and %s is not: the success path is wired and "
                "declines go nowhere." % (PI_OK, PI_FAIL))
    if billing_gap:
        return ("billing-gap",
                "the account has active subscriptions and %s is not subscribed. "
                "Renewal declines and exhausted retries are invisible." % INV_FAIL)
    if PI_FAIL in events or INV_FAIL in events:
        return ("covered", "both applicable failure events are subscribed")
    return ("no-payment-events",
            "nothing subscribes to payment success or failure at all. The gap "
            "here is the endpoint configuration rather than one event type.")


def get(session, path, **params):
    r = session.get(API + path, params=params, timeout=30)
    if r.status_code == 401:
        raise SystemExit("401 from Stripe: the key is wrong, or is for the other mode")
    r.raise_for_status()
    return r.json()


def subscribed_union(session):
    """Every event type any endpoint on this account asks for."""
    union = set()
    for ep in get(session, "/webhook_endpoints", limit=100).get("data", []):
        union.update(ep.get("enabled_events") or [])
    return union


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-events", type=int, default=100,
                    help="how many failure events to count when sizing the gap")
    args = ap.parse_args()

    key = os.environ.get("STRIPE_API_KEY")
    if not key:
        log.error("set STRIPE_API_KEY (use a restricted, read-only key)")
        return 2

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + key})

    union = subscribed_union(s)
    active = get(s, "/subscriptions", limit=1, status="active").get("data", [])
    failures = get(s, "/events", limit=args.max_events,
                   **{"types[]": [INV_FAIL]}).get("data", [])

    state, detail = verdict(union, bool(active), len(failures))
    line = "%-17s %s" % (state, detail)
    if state in ("covered", "wildcard"):
        log.info(line)
        return 0

    log.warning(line)
    log.warning("  repair: add enabled_events[]=%s and enabled_events[]=%s to an "
                "existing endpoint at %s/webhook_endpoints/{id}", PI_FAIL, INV_FAIL, API)
    log.warning("  add invoice.payment_action_required as well if renewals use 3D Secure")
    return 1


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

/**
 * Report whether anything subscribes to Stripe payment and invoice failure events.
 *
 * Read only. Three GETs, no writes: give this a RESTRICTED key with read access
 * to Webhook Endpoints, Subscriptions and Events. The repair is printed, never
 * performed.
 */
const API = 'https://api.stripe.com/v1';

const PI_OK = 'payment_intent.succeeded';
const PI_FAIL = 'payment_intent.payment_failed';
const INV_FAIL = 'invoice.payment_failed';

/**
 * Classify payment-failure coverage across both surfaces. Pure and testable.
 * The billing surface only counts when the account actually has recurring
 * billing, and failures already seen turn a gap into an incident.
 */
export function verdict(subscribed, hasActiveSubscriptions, failedInvoices) {
  const events = new Set(subscribed ?? []);
  if (events.has('*')) {
    return ['wildcard',
      'a wildcard covers both failure events, and every other type along with them.'];
  }

  const oneOffGap = events.has(PI_OK) && !events.has(PI_FAIL);
  const billingGap = Boolean(hasActiveSubscriptions) && !events.has(INV_FAIL);

  if (billingGap && failedInvoices) {
    return ['blind',
      `${failedInvoices} invoice payment(s) already failed and ${INV_FAIL} is ` +
      'not subscribed. Dunning is running right now and nothing is being told.'];
  }
  if (oneOffGap && billingGap) {
    return ['exposed',
      `neither ${PI_FAIL} nor ${INV_FAIL} is subscribed. Both the one-off and ` +
      'the billing failure paths are silent.'];
  }
  if (oneOffGap) {
    return ['one-sided',
      `${PI_OK} is subscribed and ${PI_FAIL} is not: the success path is wired ` +
      'and declines go nowhere.'];
  }
  if (billingGap) {
    return ['billing-gap',
      `the account has active subscriptions and ${INV_FAIL} is not subscribed. ` +
      'Renewal declines and exhausted retries are invisible.'];
  }
  if (events.has(PI_FAIL) || events.has(INV_FAIL)) {
    return ['covered', 'both applicable failure events are subscribed'];
  }
  return ['no-payment-events',
    'nothing subscribes to payment success or failure at all. The gap here is ' +
    'the endpoint configuration rather than one event type.'];
}

async function get(key, path, params = {}) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) {
    if (Array.isArray(v)) for (const item of v) url.searchParams.append(k, item);
    else url.searchParams.set(k, v);
  }
  const res = await fetch(url, { headers: { Authorization: `Bearer ${key}` } });
  if (res.status === 401) {
    throw new Error('401 from Stripe: the key is wrong, or is for the other mode');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

export async function subscribedUnion(key) {
  const union = new Set();
  const { data = [] } = await get(key, '/webhook_endpoints', { limit: 100 });
  for (const ep of data) for (const t of ep.enabled_events ?? []) union.add(t);
  return union;
}

async function main() {
  const key = process.env.STRIPE_API_KEY;
  if (!key) {
    console.error('set STRIPE_API_KEY (use a restricted, read-only key)');
    process.exitCode = 2;
    return;
  }

  const union = await subscribedUnion(key);
  const { data: active = [] } = await get(key, '/subscriptions',
    { limit: 1, status: 'active' });
  const { data: failures = [] } = await get(key, '/events',
    { limit: 100, 'types[]': [INV_FAIL] });

  const [state, detail] = verdict(union, active.length > 0, failures.length);
  const line = `${state.padEnd(17)} ${detail}`;
  if (state === 'covered' || state === 'wildcard') {
    console.log(line);
    return;
  }

  console.warn(line);
  console.warn(`  repair: add enabled_events[]=${PI_FAIL} and ` +
               `enabled_events[]=${INV_FAIL} to an existing endpoint at ` +
               `${API}/webhook_endpoints/{id}`);
  console.warn('  add invoice.payment_action_required as well if renewals use 3D Secure');
  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 key, 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 stripe_payment_failure_events import verdict


def test_success_subscribed_without_the_failure_is_one_sided():
    state, detail = verdict(["payment_intent.succeeded"], False, 0)
    assert state == "one-sided"
    assert "payment_intent.payment_failed" in detail


def test_active_subscriptions_with_failures_already_seen_is_an_incident():
    # Same missing subscription as the gap below; the failures make it live.
    state, detail = verdict(["payment_intent.succeeded",
                             "payment_intent.payment_failed"], True, 9)
    assert state == "blind"
    assert "9 invoice" in detail


def test_no_subscriptions_means_the_invoice_event_is_not_required():
    state, _ = verdict(["payment_intent.succeeded",
                        "payment_intent.payment_failed"], False, 0)
    assert state == "covered"


def test_both_surfaces_missing_is_reported_as_one_finding():
    state, _ = verdict(["payment_intent.succeeded", "invoice.paid"], True, 0)
    assert state == "exposed"


def test_an_account_with_no_payment_events_at_all():
    state, _ = verdict(["customer.created"], False, 0)
    assert state == "no-payment-events"


def test_a_wildcard_covers_both_surfaces():
    state, _ = verdict(["*"], True, 40)
    assert state == "wildcard"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict } from './stripe-payment-failure-events.mjs';

test('success subscribed without the failure is one sided', () => {
  const [state, detail] = verdict(['payment_intent.succeeded'], false, 0);
  assert.equal(state, 'one-sided');
  assert.match(detail, /payment_intent\.payment_failed/);
});

test('active subscriptions with failures already seen is an incident', () => {
  const [state, detail] = verdict(
    ['payment_intent.succeeded', 'payment_intent.payment_failed'], true, 9);
  assert.equal(state, 'blind');
  assert.match(detail, /9 invoice/);
});

test('no subscriptions means the invoice event is not required', () => {
  const [state] = verdict(
    ['payment_intent.succeeded', 'payment_intent.payment_failed'], false, 0);
  assert.equal(state, 'covered');
});

test('both surfaces missing is reported as one finding', () => {
  const [state] = verdict(['payment_intent.succeeded', 'invoice.paid'], true, 0);
  assert.equal(state, 'exposed');
});

test('an account with no payment events at all', () => {
  assert.equal(verdict(['customer.created'], false, 0)[0], 'no-payment-events');
});

test('a wildcard covers both surfaces', () => {
  assert.equal(verdict(['*'], true, 40)[0], 'wildcard');
});