a Connect platform has no endpoint for connected accounts

"""Report a Connect platform with no webhook destination scoped to its accounts.

Read only. Two GET requests and no writes: give this a RESTRICTED key with read
access to Webhook Endpoints and Connected accounts. 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_connect_webhook_scope")

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

# The two events that only ever come from a connected account. The endpoint
# object does not return whether it was created with connect=true, so a
# subscription to one of these is the closest thing to evidence the API offers:
# their absence proves a gap, their presence only shows somebody meant to.
CONNECT_SIGNALS = ("account.updated", "account.application.deauthorized")


def coverage(endpoints, is_platform):
    """Decide whether connected-account events have anywhere to go. Pure.

    Takes the raw /v1/webhook_endpoints list and whether this account has any
    connected accounts. Returns (state, detail). `inconclusive` is a real answer
    here and not a failure of the check: a wildcard endpoint would receive these
    events if it were Connect scoped, and nothing in the API says whether it is.
    """
    if not is_platform:
        return ("not-a-platform",
                "no connected accounts on this key, so there is no Connect traffic "
                "to scope a destination for")

    enabled = [e for e in endpoints if e.get("status") == "enabled"]
    disabled = len(endpoints) - len(enabled)

    if not enabled:
        return ("no-endpoints",
                "no enabled endpoint in this mode at all (%d disabled): nothing is "
                "being delivered anywhere, connected or otherwise" % disabled)

    subscribed = set()
    wildcards = []
    for e in enabled:
        types = e.get("enabled_events") or []
        if "*" in types:
            wildcards.append(e.get("url") or e.get("id") or "?")
        subscribed.update(types)

    have = [s for s in CONNECT_SIGNALS if s in subscribed]

    if len(have) == len(CONNECT_SIGNALS):
        return ("covered",
                "an enabled endpoint subscribes to %s" % " and ".join(CONNECT_SIGNALS))

    if wildcards and not have:
        return ("inconclusive",
                "%d endpoint(s) subscribe to * and the endpoint object never returns "
                "whether they are Connect scoped: open %s in Workbench and read "
                "whether it listens to your account or to connected accounts"
                % (len(wildcards), wildcards[0]))

    if have:
        missing = [s for s in CONNECT_SIGNALS if s not in subscribed]
        return ("thin",
                "%s is subscribed but %s is not: %s"
                % (have[0], missing[0],
                   "sellers who disconnect keep looking active"
                   if missing[0] == "account.application.deauthorized"
                   else "you see the departures and none of the verification failures"))

    tail = ", and %d disabled endpoint(s) were ignored" % disabled if disabled else ""
    return ("uncovered",
            "no enabled endpoint subscribes to %s%s"
            % (" or ".join(CONNECT_SIGNALS), tail))


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 endpoints(session):
    """Return every webhook endpoint, paginated. Usually one page; not always."""
    out = []
    params = {"limit": 100}
    while True:
        page = get(session, "/webhook_endpoints", **params)
        data = page.get("data", [])
        out.extend(data)
        if not data or not page.get("has_more"):
            return out
        params["starting_after"] = data[-1]["id"]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--show-endpoints", action="store_true",
                    help="print every endpoint with its status and subscription count")
    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})

    eps = endpoints(s)
    is_platform = bool(get(s, "/accounts", limit=1).get("data"))

    if args.show_endpoints:
        for e in eps:
            log.info("%s  %-8s %3d event type(s)  %s",
                     e.get("id", "we_?"), e.get("status", "?"),
                     len(e.get("enabled_events") or []), e.get("url", ""))

    state, detail = coverage(eps, is_platform)
    log.info("%d endpoint(s), %s: %s",
             len(eps), "platform with connected accounts" if is_platform
             else "no connected accounts", state)

    if state in ("covered", "not-a-platform"):
        log.info("  %s", detail)
        return 0

    log.warning("  %s", detail)
    log.warning("  repair: create a second destination scoped to connected accounts:")
    log.warning("  POST %s/webhook_endpoints with connect=true, "
                "url=https://<yourdomain>/stripe/connect-webhook", API)
    log.warning("  enabled_events[]=account.updated "
                "enabled_events[]=account.application.deauthorized "
                "enabled_events[]=capability.updated "
                "enabled_events[]=person.updated "
                "enabled_events[]=payout.failed")
    log.warning("  in Workbench: Create an event destination, then Connected accounts")
    log.warning("  then: read the top-level account property on each event and make "
                "any follow-up call as that account")
    return 1


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

/**
 * Report a Connect platform with no webhook destination scoped to its accounts.
 *
 * Read only. Two GET requests and no writes: give this a RESTRICTED key with
 * read access to Webhook Endpoints and Connected accounts. The repair is
 * printed, never performed.
 */
const API = 'https://api.stripe.com/v1';

// The two events that only ever come from a connected account. The endpoint
// object does not return whether it was created with connect=true, so their
// absence proves a gap and their presence only shows somebody meant to.
const CONNECT_SIGNALS = ['account.updated', 'account.application.deauthorized'];

/**
 * Decide whether connected-account events have anywhere to go. Pure.
 * Returns [state, detail]. `inconclusive` is a real answer, not a failure.
 */
export function coverage(endpoints, isPlatform) {
  if (!isPlatform) {
    return ['not-a-platform',
      'no connected accounts on this key, so there is no Connect traffic to ' +
      'scope a destination for'];
  }

  const enabled = endpoints.filter((e) => e.status === 'enabled');
  const disabled = endpoints.length - enabled.length;

  if (enabled.length === 0) {
    return ['no-endpoints',
      `no enabled endpoint in this mode at all (${disabled} disabled): nothing ` +
      'is being delivered anywhere, connected or otherwise'];
  }

  const subscribed = new Set();
  const wildcards = [];
  for (const e of enabled) {
    const types = e.enabled_events ?? [];
    if (types.includes('*')) wildcards.push(e.url ?? e.id ?? '?');
    for (const t of types) subscribed.add(t);
  }

  const have = CONNECT_SIGNALS.filter((s) => subscribed.has(s));

  if (have.length === CONNECT_SIGNALS.length) {
    return ['covered', `an enabled endpoint subscribes to ${CONNECT_SIGNALS.join(' and ')}`];
  }

  if (wildcards.length && have.length === 0) {
    return ['inconclusive',
      `${wildcards.length} endpoint(s) subscribe to * and the endpoint object ` +
      'never returns whether they are Connect scoped: open ' +
      `${wildcards[0]} in Workbench and read whether it listens to your account ` +
      'or to connected accounts'];
  }

  if (have.length) {
    const missing = CONNECT_SIGNALS.filter((s) => !subscribed.has(s));
    const consequence = missing[0] === 'account.application.deauthorized'
      ? 'sellers who disconnect keep looking active'
      : 'you see the departures and none of the verification failures';
    return ['thin', `${have[0]} is subscribed but ${missing[0]} is not: ${consequence}`];
  }

  const tail = disabled ? `, and ${disabled} disabled endpoint(s) were ignored` : '';
  return ['uncovered',
    `no enabled endpoint subscribes to ${CONNECT_SIGNALS.join(' or ')}${tail}`];
}

async function get(key, path, params = {}) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) 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();
}

async function endpoints(key) {
  const out = [];
  const params = { limit: 100 };
  for (;;) {
    const page = await get(key, '/webhook_endpoints', params);
    const data = page.data ?? [];
    out.push(...data);
    if (data.length === 0 || !page.has_more) return out;
    params.starting_after = data[data.length - 1].id;
  }
}

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 eps = await endpoints(key);
  const accounts = await get(key, '/accounts', { limit: 1 });
  const isPlatform = (accounts.data ?? []).length > 0;

  const [state, detail] = coverage(eps, isPlatform);
  console.log(`${eps.length} endpoint(s), ` +
    `${isPlatform ? 'platform with connected accounts' : 'no connected accounts'}: ${state}`);

  if (state === 'covered' || state === 'not-a-platform') {
    console.log(`  ${detail}`);
    return;
  }

  console.warn(`  ${detail}`);
  console.warn('  repair: create a second destination scoped to connected accounts:');
  console.warn(`  POST ${API}/webhook_endpoints with connect=true, ` +
               'url=https://<yourdomain>/stripe/connect-webhook');
  console.warn('  enabled_events[]=account.updated ' +
               'enabled_events[]=account.application.deauthorized ' +
               'enabled_events[]=capability.updated ' +
               'enabled_events[]=person.updated enabled_events[]=payout.failed');
  console.warn('  in Workbench: Create an event destination, then Connected accounts');
  console.warn('  then: read the top-level account property on each event and make ' +
               'any follow-up call as that account');
  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_connect_webhook_scope import coverage


def endpoint(events, status="enabled", url="https://example.com/hook"):
    return {"id": "we_1", "status": status, "url": url, "enabled_events": events}


def test_a_plain_account_is_not_asked_about_connect_scope():
    state, detail = coverage([endpoint(["charge.succeeded"])], False)
    assert state == "not-a-platform"
    assert "no connected accounts" in detail


def test_both_connect_signals_present_is_covered():
    state, _ = coverage(
        [endpoint(["account.updated", "account.application.deauthorized"])], True)
    assert state == "covered"


def test_neither_signal_is_uncovered():
    state, detail = coverage([endpoint(["charge.succeeded", "payout.paid"])], True)
    assert state == "uncovered"
    assert "account.updated" in detail


def test_a_disabled_endpoint_does_not_count_as_coverage():
    # A disabled endpoint delivers nothing, so it is the same as not having one.
    # It still gets mentioned, because "we do have that endpoint" is the first
    # thing anybody says.
    state, detail = coverage(
        [endpoint(["charge.succeeded"]),
         endpoint(["account.updated", "account.application.deauthorized"],
                  status="disabled")], True)
    assert state == "uncovered"
    assert "1 disabled endpoint(s) were ignored" in detail


def test_a_wildcard_is_inconclusive_rather_than_covered():
    # A wildcard endpoint would receive these events if it were Connect scoped,
    # and the object does not say whether it is. Reporting that honestly is the
    # whole point.
    state, detail = coverage([endpoint(["*"])], True)
    assert state == "inconclusive"
    assert "Workbench" in detail


def test_account_updated_without_deauthorized_is_half_a_subscription():
    state, detail = coverage([endpoint(["account.updated"])], True)
    assert state == "thin"
    assert "disconnect" in detail


def test_no_enabled_endpoint_at_all_says_so_first():
    state, detail = coverage([endpoint(["*"], status="disabled")], True)
    assert state == "no-endpoints"
    assert "nothing is being delivered anywhere" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { coverage } from './stripe-connect-webhook-scope.mjs';

const endpoint = (events, status = 'enabled') => ({
  id: 'we_1', status, url: 'https://example.com/hook', enabled_events: events,
});

test('a plain account is not asked about connect scope', () => {
  const [state, detail] = coverage([endpoint(['charge.succeeded'])], false);
  assert.equal(state, 'not-a-platform');
  assert.match(detail, /no connected accounts/);
});

test('both connect signals present is covered', () => {
  const [state] = coverage(
    [endpoint(['account.updated', 'account.application.deauthorized'])], true);
  assert.equal(state, 'covered');
});

test('neither signal is uncovered', () => {
  const [state, detail] = coverage([endpoint(['charge.succeeded', 'payout.paid'])], true);
  assert.equal(state, 'uncovered');
  assert.match(detail, /account.updated/);
});

test('a disabled endpoint does not count as coverage', () => {
  // A disabled endpoint delivers nothing, so it is the same as not having one.
  const [state, detail] = coverage(
    [endpoint(['charge.succeeded']),
      endpoint(['account.updated', 'account.application.deauthorized'], 'disabled')], true);
  assert.equal(state, 'uncovered');
  assert.match(detail, /1 disabled endpoint\(s\) were ignored/);
});

test('a wildcard is inconclusive rather than covered', () => {
  const [state, detail] = coverage([endpoint(['*'])], true);
  assert.equal(state, 'inconclusive');
  assert.match(detail, /Workbench/);
});

test('account.updated without deauthorized is half a subscription', () => {
  const [state, detail] = coverage([endpoint(['account.updated'])], true);
  assert.equal(state, 'thin');
  assert.match(detail, /disconnect/);
});

test('no enabled endpoint at all says so first', () => {
  const [state, detail] = coverage([endpoint(['*'], 'disabled')], true);
  assert.equal(state, 'no-endpoints');
  assert.match(detail, /nothing is being delivered anywhere/);
});