saved cards expire within 60 days and nothing warns anyone

"""Report saved Stripe cards expiring within the next 60 days.

Read only. GETs only, no writes: give this a RESTRICTED key with read access to
Subscriptions, Customers and PaymentMethods. The repair is printed, never
performed, because this script holds a credential to a live payments account.
"""
import argparse
import datetime as dt
import logging
import os
import sys
import time

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("stripe_card_expiry_window")

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

WINDOW_DAYS = 60      # how far ahead to look
NUDGE_DAYS = 45       # when to send the first email
# Wallet-backed credentials are network tokens: the issuer reissues them along
# with the card, so their printed expiry date is not a churn event.
TOKENISED_WALLETS = ("apple_pay", "google_pay", "link", "samsung_pay")


def expires_at(exp_month, exp_year):
    """Unix seconds at which a card stops being valid.

    A card is good through the END of its expiry month, so the instant it dies is
    the start of the following month. December rolls the year over explicitly
    rather than relying on month 13 normalising the way you hope.
    """
    month, year = int(exp_month), int(exp_year)
    if month == 12:
        month, year = 1, year + 1
    else:
        month += 1
    return int(dt.datetime(year, month, 1, tzinfo=dt.timezone.utc).timestamp())


def verdict(days_left, is_default=False, wallet=None):
    """Classify one saved card. Pure, so the boundaries can be tested offline.

    `days_left` is None when the PaymentMethod carries no usable expiry.
    Returns (state, detail).
    """
    if days_left is None:
        return ("unreadable", "no exp_month/exp_year on this payment method")
    if days_left <= 0:
        return ("expired",
                "already expired%s; this is a decline that has happened, not one "
                "coming" % (" and it is the billing default" if is_default else ""))
    if wallet in TOKENISED_WALLETS:
        return ("tokenised",
                "prints an expiry in %.0f day(s) but is a %s credential, which is "
                "reissued with the card. Do not email this customer."
                % (days_left, wallet))
    if days_left > WINDOW_DAYS:
        return ("ok", "expires in %.0f day(s), outside the %d day window"
                % (days_left, WINDOW_DAYS))
    if is_default:
        return ("urgent",
                "expires in %.0f day(s) and is the billing default: name the renewal "
                "that fails and email the portal link today" % days_left)
    return ("warn",
            "expires in %.0f day(s); the nudge belongs at %d days, before the "
            "decline rather than after it" % (days_left, NUDGE_DAYS))


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 page_all(session, path, limit, **params):
    """Yield objects from a paginated list endpoint until `limit` is reached."""
    seen = 0
    while True:
        page = get(session, path, **params)
        data = page.get("data", [])
        for obj in data:
            yield obj
            seen += 1
        if not data or not page.get("has_more") or seen >= limit:
            return
        params["starting_after"] = data[-1]["id"]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-subscriptions", type=int, default=1000,
                    help="stop after this many active subscriptions")
    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})

    now = time.time()
    flagged = 0
    seen_customers = set()
    for sub in page_all(s, "/subscriptions", args.max_subscriptions,
                        status="active", limit=100):
        cid = sub.get("customer")
        if not cid or cid in seen_customers:
            continue
        seen_customers.add(cid)

        customer = get(s, "/customers/" + cid)
        invoice_settings = customer.get("invoice_settings") or {}
        defaults = {sub.get("default_payment_method"),
                    invoice_settings.get("default_payment_method")}
        defaults.discard(None)

        pms = get(s, "/payment_methods", customer=cid, type="card", limit=100)
        for pm in pms.get("data", []):
            card = pm.get("card") or {}
            if not card.get("exp_month") or not card.get("exp_year"):
                days_left = None
            else:
                days_left = (expires_at(card["exp_month"], card["exp_year"]) - now) / 86400.0
            wallet = (card.get("wallet") or {}).get("type")
            state, detail = verdict(days_left, pm.get("id") in defaults, wallet)

            line = "%-10s %s  %s  %s" % (state, cid, pm.get("id"), detail)
            if state in ("ok", "tokenised"):
                log.info(line)
                continue
            log.warning(line)
            flagged += 1

    if not flagged:
        log.info("clear      no card on an active subscription expires within %d days",
                 WINDOW_DAYS)
        return 0

    log.warning("  %d card(s) need a nudge. Email a portal link, do not wait for "
                "the decline:", flagged)
    log.warning("  POST %s/billing_portal/sessions -d customer=cus_X "
                "-d return_url=https://example.com/billing", API)
    log.warning("  and turn on Smart Retries at "
                "https://dashboard.stripe.com/settings/billing/automatic")
    return 1


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

/**
 * Report saved Stripe cards expiring within the next 60 days.
 *
 * Read only. GETs only, no writes: give this a RESTRICTED key with read access
 * to Subscriptions, Customers and PaymentMethods. The repair is printed only.
 */
const API = 'https://api.stripe.com/v1';

export const WINDOW_DAYS = 60; // how far ahead to look
const NUDGE_DAYS = 45;         // when to send the first email
// Wallet-backed credentials are network tokens: the issuer reissues them along
// with the card, so their printed expiry date is not a churn event.
const TOKENISED_WALLETS = new Set(['apple_pay', 'google_pay', 'link', 'samsung_pay']);

/**
 * Unix seconds at which a card stops being valid: the start of the month after
 * its expiry month, with December rolled into the next year explicitly.
 */
export function expiresAt(expMonth, expYear) {
  let month = Number(expMonth);
  let year = Number(expYear);
  if (month === 12) { month = 1; year += 1; } else { month += 1; }
  return Date.UTC(year, month - 1, 1) / 1000;
}

/**
 * Classify one saved card. Pure, so the boundaries can be tested offline.
 * `daysLeft` is null when the PaymentMethod carries no usable expiry.
 */
export function verdict(daysLeft, isDefault = false, wallet = null) {
  if (daysLeft === null || daysLeft === undefined) {
    return ['unreadable', 'no exp_month/exp_year on this payment method'];
  }
  if (daysLeft <= 0) {
    return ['expired',
      `already expired${isDefault ? ' and it is the billing default' : ''}; ` +
      'this is a decline that has happened, not one coming'];
  }
  if (TOKENISED_WALLETS.has(wallet)) {
    return ['tokenised',
      `prints an expiry in ${daysLeft.toFixed(0)} day(s) but is a ${wallet} ` +
      'credential, which is reissued with the card. Do not email this customer.'];
  }
  if (daysLeft > WINDOW_DAYS) {
    return ['ok',
      `expires in ${daysLeft.toFixed(0)} day(s), outside the ${WINDOW_DAYS} day window`];
  }
  if (isDefault) {
    return ['urgent',
      `expires in ${daysLeft.toFixed(0)} day(s) and is the billing default: name ` +
      'the renewal that fails and email the portal link today'];
  }
  return ['warn',
    `expires in ${daysLeft.toFixed(0)} day(s); the nudge belongs at ${NUDGE_DAYS} ` +
    'days, before the decline rather than after it'];
}

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();
}

export async function pageAll(key, path, limit, params = {}) {
  const out = [];
  const q = { limit: 100, ...params };
  for (;;) {
    const page = await get(key, path, q);
    const data = page.data ?? [];
    out.push(...data);
    if (data.length === 0 || !page.has_more || out.length >= limit) return out;
    q.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 now = Date.now() / 1000;
  let flagged = 0;
  const seen = new Set();
  for (const sub of await pageAll(key, '/subscriptions', 1000, { status: 'active' })) {
    const cid = sub.customer;
    if (!cid || seen.has(cid)) continue;
    seen.add(cid);

    const customer = await get(key, `/customers/${cid}`);
    const defaults = new Set([
      sub.default_payment_method,
      customer.invoice_settings?.default_payment_method,
    ].filter(Boolean));

    const pms = await get(key, '/payment_methods', { customer: cid, type: 'card', limit: 100 });
    for (const pm of pms.data ?? []) {
      const card = pm.card ?? {};
      const daysLeft = (card.exp_month && card.exp_year)
        ? (expiresAt(card.exp_month, card.exp_year) - now) / 86400
        : null;
      const [state, detail] = verdict(daysLeft, defaults.has(pm.id), card.wallet?.type ?? null);

      const line = `${state.padEnd(10)} ${cid}  ${pm.id}  ${detail}`;
      if (state === 'ok' || state === 'tokenised') { console.log(line); continue; }
      console.warn(line);
      flagged += 1;
    }
  }

  if (!flagged) {
    console.log(`clear      no card on an active subscription expires within ${WINDOW_DAYS} days`);
    return;
  }

  console.warn(`  ${flagged} card(s) need a nudge. Email a portal link, do not wait for the decline:`);
  console.warn(`  POST ${API}/billing_portal/sessions -d customer=cus_X -d return_url=https://example.com/billing`);
  console.warn('  and turn on Smart Retries at https://dashboard.stripe.com/settings/billing/automatic');
  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; });
}

import datetime as dt

from stripe_card_expiry_window import expires_at, verdict


def utc(year, month):
    return int(dt.datetime(year, month, 1, tzinfo=dt.timezone.utc).timestamp())


def test_a_card_is_valid_through_the_end_of_its_month():
    # 04/2029 dies at the first instant of May, not of April.
    assert expires_at(4, 2029) == utc(2029, 5)


def test_december_rolls_into_the_next_year():
    assert expires_at(12, 2026) == utc(2027, 1)


def test_february_of_a_leap_year_still_lands_on_march():
    assert expires_at(2, 2028) == utc(2028, 3)


def test_an_expiry_already_past_is_a_decline_that_happened():
    state, detail = verdict(-3.0, is_default=True)
    assert state == "expired"
    assert "billing default" in detail


def test_the_window_edge_is_inclusive():
    assert verdict(60.0)[0] == "warn"
    assert verdict(60.1)[0] == "ok"


def test_the_default_card_is_its_own_bucket():
    assert verdict(20.0)[0] == "warn"
    assert verdict(20.0, is_default=True)[0] == "urgent"


def test_wallet_credentials_are_not_warned_about():
    state, detail = verdict(10.0, is_default=True, wallet="apple_pay")
    assert state == "tokenised"
    assert "reissued" in detail


def test_a_card_with_no_expiry_is_not_silently_fine():
    assert verdict(None)[0] == "unreadable"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { expiresAt, verdict } from './stripe-card-expiry-window.mjs';

test('a card is valid through the end of its month', () => {
  assert.equal(expiresAt(4, 2029), Date.UTC(2029, 4, 1) / 1000);
});

test('december rolls into the next year', () => {
  assert.equal(expiresAt(12, 2026), Date.UTC(2027, 0, 1) / 1000);
});

test('february of a leap year still lands on march', () => {
  assert.equal(expiresAt(2, 2028), Date.UTC(2028, 2, 1) / 1000);
});

test('an expiry already past is a decline that happened', () => {
  const [state, detail] = verdict(-3.0, true);
  assert.equal(state, 'expired');
  assert.match(detail, /billing default/);
});

test('the window edge is inclusive', () => {
  assert.equal(verdict(60.0)[0], 'warn');
  assert.equal(verdict(60.1)[0], 'ok');
});

test('the default card is its own bucket', () => {
  assert.equal(verdict(20.0)[0], 'warn');
  assert.equal(verdict(20.0, true)[0], 'urgent');
});

test('wallet credentials are not warned about', () => {
  const [state, detail] = verdict(10.0, true, 'apple_pay');
  assert.equal(state, 'tokenised');
  assert.match(detail, /reissued/);
});

test('a card with no expiry is not silently fine', () => {
  assert.equal(verdict(null)[0], 'unreadable');
});