the endpoint listens for charge.succeeded, not payment_intent

"""Report webhook endpoints subscribed to charge events on a PaymentIntent integration.

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

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

CHARGE = "charge.succeeded"
INTENT = "payment_intent.succeeded"
SESSION = "checkout.session.completed"


def verdict(enabled_events, fired_types):
    """Classify one endpoint's fulfilment subscription. Pure and testable.

    `fired_types` is the set of modern success types actually seen in the
    retained window. Checked in order of how much fulfilment they cost you:
    a missing path first, then a duplicated one. Returns (state, detail).
    """
    events = set(enabled_events or [])
    fired = set(fired_types or [])
    modern = INTENT in events or SESSION in events

    if "*" in events:
        return ("wildcard",
                "a wildcard delivers both shapes of the same payment. Fulfilment "
                "has to pick one and ignore the other, explicitly.")

    if CHARGE in events and not modern:
        if fired & {INTENT, SESSION}:
            return ("stale",
                    "%s is the only success subscription, but %s fired in the "
                    "retained window. The Charge carries neither the intent "
                    "metadata nor client_reference_id."
                    % (CHARGE, ", ".join(sorted(fired & {INTENT, SESSION}))))
        return ("legacy",
                "%s only, and no PaymentIntent or Checkout events fired. This "
                "looks like a genuine Charges API integration rather than a "
                "stale subscription." % CHARGE)

    if SESSION in fired and SESSION not in events:
        return ("checkout-gap",
                "Checkout Sessions are completing and %s is not subscribed. No "
                "charge or payment_intent subscription implies it." % SESSION)

    if CHARGE in events and modern:
        return ("overlapping",
                "%s and the fulfilment event are both subscribed, so one payment "
                "arrives twice in two shapes. Fulfil on one and drop the other."
                % CHARGE)

    return ("aligned", "fulfilment events match the integration")


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 modern_types_seen(session):
    """Which of the modern success types have fired in the retained window."""
    page = get(session, "/events", limit=100, **{"types[]": [INTENT, SESSION]})
    return {ev.get("type") for ev in page.get("data", [])}


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--quiet-aligned", action="store_true",
                    help="print only the endpoints that need attention")
    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", [])
    if not endpoints:
        log.info("no webhook endpoints configured for this key's mode")
        return 0

    fired = modern_types_seen(s)
    log.info("modern success types seen: %s", ", ".join(sorted(fired)) or "none")

    bad = 0
    for ep in endpoints:
        state, detail = verdict(ep.get("enabled_events"), fired)
        line = "%-12s %s  %s" % (state, ep.get("url", "?"), detail)
        if state in ("aligned", "legacy"):
            if not args.quiet_aligned:
                log.info(line)
            continue
        bad += 1
        log.warning(line)
        want = SESSION if SESSION in fired else INTENT
        log.warning("  repair: add enabled_events[]=%s to %s/webhook_endpoints/%s, "
                    "ship the handler branch, then remove %s",
                    want, API, ep["id"], CHARGE)

    log.info("%d endpoint(s), %d needing attention", len(endpoints), bad)
    return 1 if bad else 0


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

/**
 * Report webhook endpoints subscribed to charge events on a PaymentIntent integration.
 *
 * 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 CHARGE = 'charge.succeeded';
const INTENT = 'payment_intent.succeeded';
const SESSION = 'checkout.session.completed';

/**
 * Classify one endpoint's fulfilment subscription. Pure and testable.
 * Checked in order of how much fulfilment they cost you: a missing path first,
 * then a duplicated one.
 */
export function verdict(enabledEvents, firedTypes) {
  const events = new Set(enabledEvents ?? []);
  const fired = new Set(firedTypes ?? []);
  const modern = events.has(INTENT) || events.has(SESSION);

  if (events.has('*')) {
    return ['wildcard',
      'a wildcard delivers both shapes of the same payment. Fulfilment has to ' +
      'pick one and ignore the other, explicitly.'];
  }

  if (events.has(CHARGE) && !modern) {
    const seen = [INTENT, SESSION].filter((t) => fired.has(t));
    if (seen.length > 0) {
      return ['stale',
        `${CHARGE} is the only success subscription, but ${seen.join(', ')} ` +
        'fired in the retained window. The Charge carries neither the intent ' +
        'metadata nor client_reference_id.'];
    }
    return ['legacy',
      `${CHARGE} only, and no PaymentIntent or Checkout events fired. This ` +
      'looks like a genuine Charges API integration rather than a stale subscription.'];
  }

  if (fired.has(SESSION) && !events.has(SESSION)) {
    return ['checkout-gap',
      `Checkout Sessions are completing and ${SESSION} is not subscribed. No ` +
      'charge or payment_intent subscription implies it.'];
  }

  if (events.has(CHARGE) && modern) {
    return ['overlapping',
      `${CHARGE} and the fulfilment event are both subscribed, so one payment ` +
      'arrives twice in two shapes. Fulfil on one and drop the other.'];
  }

  return ['aligned', 'fulfilment events match the integration'];
}

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 modernTypesSeen(key) {
  const page = await get(key, '/events', { limit: 100, 'types[]': [INTENT, SESSION] });
  return new Set((page.data ?? []).map((ev) => ev.type));
}

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 });
  if (endpoints.length === 0) {
    console.log("no webhook endpoints configured for this key's mode");
    return;
  }

  const fired = await modernTypesSeen(key);
  console.log(`modern success types seen: ${[...fired].sort().join(', ') || 'none'}`);

  let bad = 0;
  for (const ep of endpoints) {
    const [state, detail] = verdict(ep.enabled_events, fired);
    const line = `${state.padEnd(12)} ${ep.url ?? '?'}  ${detail}`;
    if (state === 'aligned' || state === 'legacy') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    const want = fired.has(SESSION) ? SESSION : INTENT;
    console.warn(`  repair: add enabled_events[]=${want} to ` +
                 `${API}/webhook_endpoints/${ep.id}, ship the handler branch, ` +
                 `then remove ${CHARGE}`);
  }

  console.log(`${endpoints.length} endpoint(s), ${bad} needing attention`);
  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 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_charge_event_drift import verdict


def test_charge_only_while_intents_fire_is_stale():
    state, detail = verdict(["charge.succeeded"], ["payment_intent.succeeded"])
    assert state == "stale"
    assert "client_reference_id" in detail


def test_the_same_config_with_no_modern_traffic_is_a_real_charges_integration():
    # Identical enabled_events. Only the account's traffic tells them apart.
    state, _ = verdict(["charge.succeeded"], [])
    assert state == "legacy"


def test_subscribing_to_both_is_double_fulfilment():
    state, detail = verdict(["charge.succeeded", "payment_intent.succeeded"],
                            ["payment_intent.succeeded"])
    assert state == "overlapping"
    assert "twice" in detail


def test_checkout_sessions_firing_with_no_session_subscription():
    state, _ = verdict(["payment_intent.succeeded"],
                       ["payment_intent.succeeded", "checkout.session.completed"])
    assert state == "checkout-gap"


def test_a_matching_subscription_is_aligned():
    state, _ = verdict(["payment_intent.succeeded", "checkout.session.completed"],
                       ["payment_intent.succeeded", "checkout.session.completed"])
    assert state == "aligned"


def test_a_wildcard_is_called_out_rather_than_passed():
    state, _ = verdict(["*"], ["payment_intent.succeeded"])
    assert state == "wildcard"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict } from './stripe-charge-event-drift.mjs';

test('charge only while intents fire is stale', () => {
  const [state, detail] = verdict(['charge.succeeded'], ['payment_intent.succeeded']);
  assert.equal(state, 'stale');
  assert.match(detail, /client_reference_id/);
});

test('the same config with no modern traffic is a real charges integration', () => {
  assert.equal(verdict(['charge.succeeded'], [])[0], 'legacy');
});

test('subscribing to both is double fulfilment', () => {
  const [state, detail] = verdict(
    ['charge.succeeded', 'payment_intent.succeeded'], ['payment_intent.succeeded']);
  assert.equal(state, 'overlapping');
  assert.match(detail, /twice/);
});

test('checkout sessions firing with no session subscription', () => {
  const [state] = verdict(['payment_intent.succeeded'],
    ['payment_intent.succeeded', 'checkout.session.completed']);
  assert.equal(state, 'checkout-gap');
});

test('a matching subscription is aligned', () => {
  const [state] = verdict(['payment_intent.succeeded', 'checkout.session.completed'],
    ['payment_intent.succeeded', 'checkout.session.completed']);
  assert.equal(state, 'aligned');
});

test('a wildcard is called out rather than passed', () => {
  assert.equal(verdict(['*'], ['payment_intent.succeeded'])[0], 'wildcard');
});