elevated-risk card charges are captured with no 3DS

"""Report Stripe card charges captured at elevated risk without 3D Secure.

Read only. One paginated GET and no writes: give this a RESTRICTED key with read
access to Charges. The Radar rules that repair this are printed, never applied,
because a rule change reprices every payment on the account and this script holds
a credential to a live one.
"""
import argparse
import collections
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_3ds_coverage")

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

# Radar levels where an unauthenticated capture is a finding rather than normal.
ELEVATED = ("elevated", "highest")

# The only result that carries the liability shift.
AUTHENTICATED = "authenticated"

# Mastercard's fraud monitoring penalises merchants at or below this share.
SHARE_FLOOR = 0.10


def classify(charge):
    """Classify one charge. Pure. Returns (state, detail).

    `unprotected` is the finding: Radar scored the payment elevated or higher,
    it was captured, and no authentication happened, so a fraud dispute on it
    has no liability shift to invoke.
    """
    pmd = charge.get("payment_method_details") or {}
    if pmd.get("type") != "card":
        return ("not_card",
                "%s, which authenticates differently or not at all"
                % (pmd.get("type") or "no payment method details",))
    if charge.get("status") != "succeeded":
        return ("not_settled",
                "status is %r, so it cannot be disputed" % (charge.get("status"),))

    risk = (charge.get("outcome") or {}).get("risk_level")
    tds = (pmd.get("card") or {}).get("three_d_secure")

    if tds is None:
        if risk in ELEVATED:
            return ("unprotected",
                    "risk_level %s captured with three_d_secure null. Radar "
                    "flagged it, nothing authenticated it, and the fraud "
                    "liability is yours." % risk)
        return ("no_3ds",
                "risk_level %s, no authentication. Ordinary, but it counts "
                "against the account 3DS share." % (risk or "unknown",))

    result = tds.get("result")
    if result == AUTHENTICATED:
        return ("protected",
                "authenticated; liability for most fraud disputes sits with the issuer")
    if risk in ELEVATED:
        return ("attempted",
                "three_d_secure.result is %r on a %s risk charge. The flow ran "
                "and the issuer did not complete it, so this looks covered and "
                "is not." % (result, risk))
    return ("attempted",
            "three_d_secure.result is %r, which is not an authentication" % (result,))


def coverage(authenticated, card_charges, floor=SHARE_FLOOR):
    """Account-wide 3DS share. Pure. Returns (state, detail).

    Only `authenticated` counts in the numerator: an acknowledged attempt is
    not an authentication, and counting it overstates the share against the
    number the card networks compute.
    """
    if not card_charges:
        return ("no_volume", "no card charges in the window")
    share = authenticated / card_charges
    if share <= floor:
        return ("low",
                "%.1f%% of card charges authenticated, at or below the %.0f%% "
                "where Mastercard fraud monitoring applies"
                % (share * 100, floor * 100))
    return ("ok", "%.1f%% of card charges authenticated" % (share * 100))


def get(session, path, params=None):
    r = session.get(API + path, params=params or {}, 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 charges(session, since, limit):
    """Yield charges created since `since`, newest first."""
    seen = 0
    params = {"limit": 100, "created[gte]": int(since)}
    while True:
        page = get(session, "/charges", params)
        data = page.get("data", [])
        for c in data:
            yield c
            seen += 1
        if not data or not page.get("has_more") or seen >= limit:
            break
        params["starting_after"] = data[-1]["id"]


REQUEST_RULE = ("Request 3D Secure if :risk_level: != 'normal' "
                "and :amount_in_usd: > 25")
BLOCK_RULE = ("Block if not :is_3d_secure: and :risk_level: != 'normal' "
              "and not :is_off_session: and :digital_wallet: != 'apple_pay' "
              "and not (:digital_wallet: = 'android_pay' and :has_cryptogram:)")


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=90, help="how far back to read charges")
    ap.add_argument("--max-charges", type=int, default=5000,
                    help="stop paginating after this many charges")
    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})

    since = time.time() - args.days * 86400
    states = collections.Counter()
    card_charges = 0
    authenticated = 0
    findings = []

    for c in charges(s, since, args.max_charges):
        state, detail = classify(c)
        states[state] += 1
        if state == "not_card":
            continue
        if state != "not_settled":
            card_charges += 1
        if state == "protected":
            authenticated += 1
        if state in ("unprotected", "attempted"):
            findings.append((c, state, detail))

    for c, state, detail in findings:
        log.warning("%-12s %s  %s %s  %s", state, c.get("id", "?"),
                    c.get("amount"), (c.get("currency") or "?").upper(), detail)

    share_state, share_detail = coverage(authenticated, card_charges)
    log.info("%d card charge(s): %d unprotected, %d attempted, %d authenticated",
             card_charges, states["unprotected"], states["attempted"], authenticated)
    if share_state == "low":
        log.warning("3DS share: %s", share_detail)
    else:
        log.info("3DS share: %s", share_detail)

    if findings or share_state == "low":
        log.warning("  repair, in Dashboard, Radar, Rules, add both together:")
        log.warning("    %s", REQUEST_RULE)
        log.warning("    %s", BLOCK_RULE)
        log.warning("  the request rule alone lets cards whose issuer will not "
                    "authenticate proceed unauthenticated anyway")
        log.warning("  note that early fraud warnings still arrive on authenticated "
                    "payments and still count toward the Visa VAMP ratio")
        return 1
    return 0


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

/**
 * Report Stripe card charges captured at elevated risk without 3D Secure.
 *
 * Read only. One paginated GET and no writes: give this a RESTRICTED key with
 * read access to Charges. The Radar rules that repair this are printed, never
 * applied, because a rule change reprices every payment on the account.
 */
const API = 'https://api.stripe.com/v1';

const ELEVATED = ['elevated', 'highest'];
const AUTHENTICATED = 'authenticated';
export const SHARE_FLOOR = 0.10;

/**
 * Classify one charge. Pure. Returns [state, detail].
 */
export function classify(charge) {
  const pmd = charge.payment_method_details ?? {};
  if (pmd.type !== 'card') {
    return ['not_card',
      `${pmd.type ?? 'no payment method details'}, which authenticates ` +
      'differently or not at all'];
  }
  if (charge.status !== 'succeeded') {
    return ['not_settled',
      `status is ${JSON.stringify(charge.status)}, so it cannot be disputed`];
  }

  const risk = (charge.outcome ?? {}).risk_level;
  const tds = (pmd.card ?? {}).three_d_secure;

  if (tds === null || tds === undefined) {
    if (ELEVATED.includes(risk)) {
      return ['unprotected',
        `risk_level ${risk} captured with three_d_secure null. Radar flagged ` +
        'it, nothing authenticated it, and the fraud liability is yours.'];
    }
    return ['no_3ds',
      `risk_level ${risk ?? 'unknown'}, no authentication. Ordinary, but it ` +
      'counts against the account 3DS share.'];
  }

  const result = tds.result;
  if (result === AUTHENTICATED) {
    return ['protected',
      'authenticated; liability for most fraud disputes sits with the issuer'];
  }
  if (ELEVATED.includes(risk)) {
    return ['attempted',
      `three_d_secure.result is ${JSON.stringify(result)} on a ${risk} risk ` +
      'charge. The flow ran and the issuer did not complete it, so this looks ' +
      'covered and is not.'];
  }
  return ['attempted',
    `three_d_secure.result is ${JSON.stringify(result)}, which is not an authentication`];
}

/**
 * Account-wide 3DS share. Pure. Returns [state, detail]. Only authenticated
 * charges count in the numerator; an acknowledged attempt is not an
 * authentication.
 */
export function coverage(authenticated, cardCharges, floor = SHARE_FLOOR) {
  if (!cardCharges) return ['no_volume', 'no card charges in the window'];
  const share = authenticated / cardCharges;
  if (share <= floor) {
    return ['low',
      `${(share * 100).toFixed(1)}% of card charges authenticated, at or below ` +
      `the ${(floor * 100).toFixed(0)}% where Mastercard fraud monitoring applies`];
  }
  return ['ok', `${(share * 100).toFixed(1)}% of card charges authenticated`];
}

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* charges(key, since, limit = 5000) {
  let seen = 0;
  const params = { limit: 100, 'created[gte]': Math.floor(since) };
  for (;;) {
    const page = await get(key, '/charges', params);
    const data = page.data ?? [];
    for (const c of data) { yield c; seen += 1; }
    if (data.length === 0 || !page.has_more || seen >= limit) break;
    params.starting_after = data[data.length - 1].id;
  }
}

const REQUEST_RULE =
  "Request 3D Secure if :risk_level: != 'normal' and :amount_in_usd: > 25";
const BLOCK_RULE =
  "Block if not :is_3d_secure: and :risk_level: != 'normal' and not " +
  ":is_off_session: and :digital_wallet: != 'apple_pay' and not " +
  "(:digital_wallet: = 'android_pay' and :has_cryptogram:)";

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 days = process.argv.includes('--days')
    ? Number(process.argv[process.argv.indexOf('--days') + 1]) : 90;
  const since = Date.now() / 1000 - days * 86400;

  const states = new Map();
  let cardCharges = 0;
  let authenticated = 0;
  const findings = [];

  for await (const c of charges(key, since)) {
    const [state, detail] = classify(c);
    states.set(state, (states.get(state) ?? 0) + 1);
    if (state === 'not_card') continue;
    if (state !== 'not_settled') cardCharges += 1;
    if (state === 'protected') authenticated += 1;
    if (state === 'unprotected' || state === 'attempted') {
      findings.push([c, state, detail]);
    }
  }

  for (const [c, state, detail] of findings) {
    console.warn(`${state.padEnd(12)} ${c.id ?? '?'}  ${c.amount} ` +
                 `${(c.currency ?? '?').toUpperCase()}  ${detail}`);
  }

  const [shareState, shareDetail] = coverage(authenticated, cardCharges);
  console.log(`${cardCharges} card charge(s): ${states.get('unprotected') ?? 0} ` +
              `unprotected, ${states.get('attempted') ?? 0} attempted, ` +
              `${authenticated} authenticated`);
  if (shareState === 'low') console.warn(`3DS share: ${shareDetail}`);
  else console.log(`3DS share: ${shareDetail}`);

  if (findings.length || shareState === 'low') {
    console.warn('  repair, in Dashboard, Radar, Rules, add both together:');
    console.warn(`    ${REQUEST_RULE}`);
    console.warn(`    ${BLOCK_RULE}`);
    console.warn('  the request rule alone lets cards whose issuer will not ' +
                 'authenticate proceed unauthenticated anyway');
    console.warn('  note that early fraud warnings still arrive on authenticated ' +
                 'payments and still count toward the Visa VAMP ratio');
    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_3ds_coverage import classify, coverage


def card_charge(risk="normal", three_d_secure=None, status="succeeded"):
    card = {"brand": "visa"}
    if three_d_secure is not None:
        card["three_d_secure"] = three_d_secure
    return {
        "id": "ch_1", "status": status, "amount": 9900, "currency": "usd",
        "outcome": {"risk_level": risk},
        "payment_method_details": {"type": "card", "card": card},
    }


def test_elevated_risk_with_no_authentication_is_the_finding():
    state, detail = classify(card_charge(risk="elevated"))
    assert state == "unprotected"
    assert "liability" in detail


def test_normal_risk_with_no_authentication_is_not_a_finding():
    # Ordinary traffic. Flagging it would bury the elevated-risk charges.
    state, detail = classify(card_charge(risk="normal"))
    assert state == "no_3ds"
    assert "share" in detail


def test_an_acknowledged_attempt_is_not_an_authentication():
    state, detail = classify(card_charge(
        risk="highest", three_d_secure={"result": "attempt_acknowledged"}))
    assert state == "attempted"
    assert "not" in detail
    assert classify(card_charge(
        risk="highest", three_d_secure={"result": "authenticated"}))[0] == "protected"


def test_non_card_and_unsettled_charges_are_out_of_scope():
    assert classify({"payment_method_details": {"type": "us_bank_account"}})[0] == "not_card"
    assert classify(card_charge(risk="highest", status="failed"))[0] == "not_settled"


def test_the_ten_percent_coverage_floor_is_inclusive():
    assert coverage(10, 100)[0] == "low"
    assert coverage(11, 100)[0] == "ok"
    assert coverage(0, 0)[0] == "no_volume"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { classify, coverage } from './stripe-3ds-coverage.mjs';

function cardCharge(risk = 'normal', threeDSecure = null, status = 'succeeded') {
  const card = { brand: 'visa' };
  if (threeDSecure !== null) card.three_d_secure = threeDSecure;
  return {
    id: 'ch_1',
    status,
    amount: 9900,
    currency: 'usd',
    outcome: { risk_level: risk },
    payment_method_details: { type: 'card', card },
  };
}

test('elevated risk with no authentication is the finding', () => {
  const [state, detail] = classify(cardCharge('elevated'));
  assert.equal(state, 'unprotected');
  assert.match(detail, /liability/);
});

test('normal risk with no authentication is not a finding', () => {
  const [state, detail] = classify(cardCharge('normal'));
  assert.equal(state, 'no_3ds');
  assert.match(detail, /share/);
});

test('an acknowledged attempt is not an authentication', () => {
  const [state, detail] = classify(
    cardCharge('highest', { result: 'attempt_acknowledged' }));
  assert.equal(state, 'attempted');
  assert.match(detail, /not/);
  assert.equal(
    classify(cardCharge('highest', { result: 'authenticated' }))[0], 'protected');
});

test('non card and unsettled charges are out of scope', () => {
  assert.equal(
    classify({ payment_method_details: { type: 'us_bank_account' } })[0], 'not_card');
  assert.equal(classify(cardCharge('highest', null, 'failed'))[0], 'not_settled');
});

test('the ten percent coverage floor is inclusive', () => {
  assert.equal(coverage(10, 100)[0], 'low');
  assert.equal(coverage(11, 100)[0], 'ok');
  assert.equal(coverage(0, 0)[0], 'no_volume');
});