live mode has no webhook endpoint, so nothing is ever pushed

"""Report whether this mode has a webhook endpoint at all, and whether it needs one.

Read only. Two GETs, no writes: give this a RESTRICTED key with read access to
Webhook Endpoints 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_live_webhook_endpoints")

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

# Events that prove this mode is carrying real business traffic. If any of these
# fired and no endpoint exists, work that should follow a payment never happened.
TRAFFIC_TYPES = ("payment_intent.succeeded", "checkout.session.completed",
                 "invoice.paid")


def verdict(endpoints, payment_events, livemode):
    """Classify webhook coverage for one mode. Pure, so the rules can be tested.

    `endpoints` is the raw data array from /v1/webhook_endpoints, `payment_events`
    the count of business events seen in the retained window, and `livemode`
    whether this key reads live data. Returns (state, detail).
    """
    eps = list(endpoints or [])
    enabled = [e for e in eps if e.get("status") == "enabled"]
    if not eps:
        if payment_events:
            return ("blind",
                    "%d payment event(s) in the retained window and no webhook "
                    "endpoint to receive them. Stripe had nowhere to push, so "
                    "nothing that should follow a payment ever ran."
                    % payment_events)
        return ("empty",
                "no webhook endpoint, and no payment events in the retained "
                "window either. Nothing has been lost yet: create the endpoint "
                "before the first real payment rather than after it.")
    if not enabled:
        return ("all-disabled",
                "%d endpoint(s) exist and every one of them is disabled, which "
                "delivers exactly as much as having none." % len(eps))
    if not livemode:
        return ("test-mode",
                "%d enabled endpoint(s), all test mode. A healthy test mode is "
                "what lets this ship: re-run with a live restricted key before "
                "concluding anything about production." % len(enabled))
    return ("covered", "%d enabled endpoint(s) in this mode" % len(enabled))


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")
    if r.status_code == 403:
        raise SystemExit("403 from Stripe: the restricted key lacks read access to "
                         + path)
    r.raise_for_status()
    return r.json()


def is_livemode(key):
    """Mode from the key prefix. Confirmed against object livemode where possible."""
    return not key.startswith(("sk_test_", "rk_test_", "pk_test_"))


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-events", type=int, default=100,
                    help="how many recent payment events to count as evidence")
    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})

    endpoints = get(s, "/webhook_endpoints", limit=100).get("data", [])
    events = get(s, "/events", limit=min(args.max_events, 100),
                 **{"types[]": list(TRAFFIC_TYPES)}).get("data", [])

    # Prefer the objects over the key prefix: a restricted key can be named
    # anything, but livemode on a returned object is the account's own answer.
    livemode = is_livemode(key)
    for obj in list(endpoints) + list(events):
        if "livemode" in obj:
            livemode = bool(obj["livemode"])
            break

    state, detail = verdict(endpoints, len(events), livemode)
    line = "%-12s %s" % (state, detail)
    if state == "covered":
        log.info(line)
        for ep in endpoints:
            log.info("  %s  %d subscribed type(s)",
                     ep.get("url", "?"), len(ep.get("enabled_events") or []))
        return 0

    log.warning(line)
    if state in ("blind", "empty"):
        log.warning("  repair: POST %s/webhook_endpoints", API)
        log.warning("    -d url=https://<your-domain>/stripe/webhook")
        log.warning("    -d enabled_events[]=payment_intent.succeeded")
        log.warning("    -d enabled_events[]=payment_intent.payment_failed")
        log.warning("  then copy the secret from the response into the server "
                    "environment: the whsec_ printed by the CLI is not it")
    if state == "blind":
        log.warning("  backfill: GET %s/charges?created[gte]=<unix> and "
                    "%s/invoices?created[gte]=<unix>, which are not retention "
                    "limited the way /v1/events is", API, API)
    return 1


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

/**
 * Report whether this mode has a webhook endpoint at all, and whether it needs one.
 *
 * Read only. Two GETs, no writes: give this a RESTRICTED key with read access to
 * Webhook Endpoints and Events. The repair is printed, never performed.
 */
const API = 'https://api.stripe.com/v1';

const TRAFFIC_TYPES = ['payment_intent.succeeded', 'checkout.session.completed',
  'invoice.paid'];

/**
 * Classify webhook coverage for one mode. Pure, so the rules can be tested.
 */
export function verdict(endpoints, paymentEvents, livemode) {
  const eps = endpoints ?? [];
  const enabled = eps.filter((e) => e.status === 'enabled');
  if (eps.length === 0) {
    if (paymentEvents) {
      return ['blind',
        `${paymentEvents} payment event(s) in the retained window and no webhook ` +
        'endpoint to receive them. Stripe had nowhere to push, so nothing that ' +
        'should follow a payment ever ran.'];
    }
    return ['empty',
      'no webhook endpoint, and no payment events in the retained window either. ' +
      'Nothing has been lost yet: create the endpoint before the first real ' +
      'payment rather than after it.'];
  }
  if (enabled.length === 0) {
    return ['all-disabled',
      `${eps.length} endpoint(s) exist and every one of them is disabled, which ` +
      'delivers exactly as much as having none.'];
  }
  if (!livemode) {
    return ['test-mode',
      `${enabled.length} enabled endpoint(s), all test mode. A healthy test mode ` +
      'is what lets this ship: re-run with a live restricted key before ' +
      'concluding anything about production.'];
  }
  return ['covered', `${enabled.length} enabled endpoint(s) in this mode`];
}

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.status === 403) {
    throw new Error(`403 from Stripe: the restricted key lacks read access to ${path}`);
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

export function isLivemode(key) {
  return !/^(sk|rk|pk)_test_/.test(key);
}

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 { data: endpoints = [] } = await get(key, '/webhook_endpoints', { limit: 100 });
  const { data: events = [] } = await get(key, '/events',
    { limit: 100, 'types[]': TRAFFIC_TYPES });

  let livemode = isLivemode(key);
  for (const obj of [...endpoints, ...events]) {
    if ('livemode' in obj) { livemode = Boolean(obj.livemode); break; }
  }

  const [state, detail] = verdict(endpoints, events.length, livemode);
  const line = `${state.padEnd(12)} ${detail}`;
  if (state === 'covered') {
    console.log(line);
    for (const ep of endpoints) {
      console.log(`  ${ep.url ?? '?'}  ${(ep.enabled_events ?? []).length} subscribed type(s)`);
    }
    return;
  }

  console.warn(line);
  if (state === 'blind' || state === 'empty') {
    console.warn(`  repair: POST ${API}/webhook_endpoints`);
    console.warn('    -d url=https://<your-domain>/stripe/webhook');
    console.warn('    -d enabled_events[]=payment_intent.succeeded');
    console.warn('    -d enabled_events[]=payment_intent.payment_failed');
    console.warn('  then copy the secret from the response into the server ' +
                 'environment: the whsec_ printed by the CLI is not it');
  }
  if (state === 'blind') {
    console.warn(`  backfill: GET ${API}/charges?created[gte]=<unix> and ` +
                 `${API}/invoices?created[gte]=<unix>, which are not retention ` +
                 'limited the way /v1/events is');
  }
  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_live_webhook_endpoints import verdict, is_livemode


def test_no_endpoints_with_payments_is_the_outage():
    state, detail = verdict([], 47, True)
    assert state == "blind"
    assert "47" in detail


def test_no_endpoints_and_no_traffic_is_a_gap_not_an_outage():
    state, detail = verdict([], 0, True)
    assert state == "empty"
    assert "before the first real payment" in detail


def test_endpoints_that_are_all_disabled_deliver_nothing():
    state, _ = verdict([{"status": "disabled"}, {"status": "disabled"}], 12, True)
    assert state == "all-disabled"


def test_a_healthy_test_mode_is_not_a_pass():
    state, detail = verdict([{"status": "enabled"}], 12, False)
    assert state == "test-mode"
    assert "live restricted key" in detail


def test_an_enabled_live_endpoint_is_covered():
    state, _ = verdict([{"status": "enabled"}], 12, True)
    assert state == "covered"


def test_missing_status_does_not_count_as_enabled():
    state, _ = verdict([{}], 0, True)
    assert state == "all-disabled"


def test_key_prefix_decides_the_mode():
    assert is_livemode("rk_live_abc") is True
    assert is_livemode("rk_test_abc") is False

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict, isLivemode } from './stripe-live-webhook-endpoints.mjs';

test('no endpoints with payments is the outage', () => {
  const [state, detail] = verdict([], 47, true);
  assert.equal(state, 'blind');
  assert.match(detail, /47/);
});

test('no endpoints and no traffic is a gap not an outage', () => {
  const [state, detail] = verdict([], 0, true);
  assert.equal(state, 'empty');
  assert.match(detail, /before the first real payment/);
});

test('endpoints that are all disabled deliver nothing', () => {
  const eps = [{ status: 'disabled' }, { status: 'disabled' }];
  assert.equal(verdict(eps, 12, true)[0], 'all-disabled');
});

test('a healthy test mode is not a pass', () => {
  const [state, detail] = verdict([{ status: 'enabled' }], 12, false);
  assert.equal(state, 'test-mode');
  assert.match(detail, /live restricted key/);
});

test('an enabled live endpoint is covered', () => {
  assert.equal(verdict([{ status: 'enabled' }], 12, true)[0], 'covered');
});

test('missing status does not count as enabled', () => {
  assert.equal(verdict([{}], 0, true)[0], 'all-disabled');
});

test('key prefix decides the mode', () => {
  assert.equal(isLivemode('rk_live_abc'), true);
  assert.equal(isLivemode('rk_test_abc'), false);
});