highest-risk charges succeed instead of being blocked

"""Report Stripe charges Radar scored as highest risk that succeeded anyway.

Read only. Paginated GETs and nothing else: give this a RESTRICTED key with read
access to Charges, Disputes and Radar. 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 time

import requests

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

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

LEAKING = ("allowed", "leaked", "uncaptured")


def verdict(risk_level, status, captured, rule):
    """Classify one charge. Pure, so the precedence rules can be tested offline.

    `rule` is the charge's outcome.rule: None, a rule id string, or the expanded
    object carrying `action` and `predicate`. Returns (state, detail).
    """
    if risk_level in (None, "not_assessed"):
        return ("not_assessed",
                "Radar never scored this charge: no Radar session reached the API")
    if risk_level != "highest":
        return ("baseline", "risk_level %s, outside the scope of this check" % risk_level)
    if status != "succeeded":
        return ("stopped", "highest risk and status %s: the block held" % status)
    if not captured:
        return ("uncaptured",
                "highest risk, authorized but not captured: cancel the payment "
                "intent before the hold is captured or expires")
    action = rule.get("action") if isinstance(rule, dict) else None
    if action == "allow":
        predicate = rule.get("predicate") or rule.get("id") or "unnamed"
        return ("allowed",
                "highest risk and captured because an allow rule matched first: %s"
                % predicate)
    return ("leaked",
            "highest risk and captured with no rule named: the built-in highest "
            "risk block rule is not in force on this account")


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(session, path, cap, **params):
    out = []
    params = dict(params)
    params["limit"] = 100
    while True:
        p = get(session, path, **params)
        data = p.get("data", [])
        out.extend(data)
        if not data or not p.get("has_more") or len(out) >= cap:
            break
        params["starting_after"] = data[-1]["id"]
    return out


def fraud_charge_ids(session, cap):
    """Charge ids carrying an early fraud warning or a dispute."""
    ids = set()
    for efw in page(session, "/radar/early_fraud_warnings", cap):
        if efw.get("charge"):
            ids.add(efw["charge"] if isinstance(efw["charge"], str)
                    else efw["charge"].get("id"))
    for dispute in page(session, "/disputes", cap):
        if dispute.get("charge"):
            ids.add(dispute["charge"] if isinstance(dispute["charge"], str)
                    else dispute["charge"].get("id"))
    return {i for i in ids if i}


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 = int(time.time() - args.days * 86400)
    charges = page(s, "/charges", args.max_charges, **{"created[gte]": since})

    leaking = []
    counts = {}
    for ch in charges:
        outcome = ch.get("outcome") or {}
        state, detail = verdict(outcome.get("risk_level"), ch.get("status"),
                                ch.get("captured"), outcome.get("rule"))
        counts[state] = counts.get(state, 0) + 1
        if state in LEAKING:
            leaking.append((ch, state, detail))

    log.info("%d charge(s): %d highest-risk captured, %d stopped",
             len(charges), counts.get("allowed", 0) + counts.get("leaked", 0),
             counts.get("stopped", 0))
    if counts.get("not_assessed"):
        log.warning("%d charge(s) were never scored by Radar. Mount Stripe.js on the "
                    "payment page, or pass radar_options[session] on server-side "
                    "confirms, before tuning any rule.",
                    counts["not_assessed"])

    if not leaking:
        return 1 if counts.get("not_assessed") else 0

    fraud = fraud_charge_ids(s, 1000)
    hits = 0
    for ch, state, detail in leaking:
        marker = ""
        if ch.get("id") in fraud:
            hits += 1
            marker = "  [early fraud warning or dispute on this charge]"
        log.warning("%-12s %s %s%s", state, ch.get("id"), detail, marker)

    log.warning("  %d of %d leaked charge(s) already carry fraud evidence",
                hits, len(leaking))
    log.warning("  guard every allow rule in Dashboard, Radar, Rules by appending "
                "and :risk_level: != 'highest' to its predicate")
    log.warning("  then confirm the built-in rule if :risk_level: = 'highest' is "
                "still enabled")
    return 1


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

/**
 * Report Stripe charges Radar scored as highest risk that succeeded anyway.
 *
 * Read only. Paginated GETs and nothing else: give this a RESTRICTED key with
 * read access to Charges, Disputes and Radar. The repair is printed, never run.
 */
const API = 'https://api.stripe.com/v1';

const LEAKING = new Set(['allowed', 'leaked', 'uncaptured']);

/**
 * Classify one charge. Pure, so the precedence rules can be tested offline.
 * `rule` is outcome.rule: null, a rule id, or the expanded rule object.
 */
export function verdict(riskLevel, status, captured, rule) {
  if (riskLevel === null || riskLevel === undefined || riskLevel === 'not_assessed') {
    return ['not_assessed',
      'Radar never scored this charge: no Radar session reached the API'];
  }
  if (riskLevel !== 'highest') {
    return ['baseline', `risk_level ${riskLevel}, outside the scope of this check`];
  }
  if (status !== 'succeeded') {
    return ['stopped', `highest risk and status ${status}: the block held`];
  }
  if (!captured) {
    return ['uncaptured',
      'highest risk, authorized but not captured: cancel the payment intent ' +
      'before the hold is captured or expires'];
  }
  const action = rule && typeof rule === 'object' ? rule.action : null;
  if (action === 'allow') {
    const predicate = (rule.predicate ?? rule.id) || 'unnamed';
    return ['allowed',
      `highest risk and captured because an allow rule matched first: ${predicate}`];
  }
  return ['leaked',
    'highest risk and captured with no rule named: the built-in highest risk ' +
    'block rule is not in force on this account'];
}

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 page(key, path, cap, params = {}) {
  const out = [];
  const q = { ...params, limit: 100 };
  for (;;) {
    const p = await get(key, path, q);
    const data = p.data ?? [];
    out.push(...data);
    if (data.length === 0 || !p.has_more || out.length >= cap) break;
    q.starting_after = data[data.length - 1].id;
  }
  return out;
}

async function fraudChargeIds(key, cap) {
  const ids = new Set();
  const idOf = (c) => (typeof c === 'string' ? c : c?.id);
  for (const efw of await page(key, '/radar/early_fraud_warnings', cap)) {
    if (efw.charge) ids.add(idOf(efw.charge));
  }
  for (const dispute of await page(key, '/disputes', cap)) {
    if (dispute.charge) ids.add(idOf(dispute.charge));
  }
  ids.delete(undefined);
  return ids;
}

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 = Number(process.env.DAYS ?? 90);
  const since = Math.floor(Date.now() / 1000 - days * 86400);
  const charges = await page(key, '/charges', 5000, { 'created[gte]': since });

  const leaking = [];
  const counts = {};
  for (const ch of charges) {
    const outcome = ch.outcome ?? {};
    const [state, detail] = verdict(outcome.risk_level, ch.status, ch.captured,
                                    outcome.rule);
    counts[state] = (counts[state] ?? 0) + 1;
    if (LEAKING.has(state)) leaking.push([ch, state, detail]);
  }

  const captured = (counts.allowed ?? 0) + (counts.leaked ?? 0);
  console.log(`${charges.length} charge(s): ${captured} highest-risk captured, ` +
              `${counts.stopped ?? 0} stopped`);
  if (counts.not_assessed) {
    console.warn(`${counts.not_assessed} charge(s) were never scored by Radar. Mount ` +
                 'Stripe.js on the payment page, or pass radar_options[session] on ' +
                 'server-side confirms, before tuning any rule.');
  }

  if (leaking.length === 0) {
    if (counts.not_assessed) process.exitCode = 1;
    return;
  }

  const fraud = await fraudChargeIds(key, 1000);
  let hits = 0;
  for (const [ch, state, detail] of leaking) {
    let marker = '';
    if (fraud.has(ch.id)) {
      hits += 1;
      marker = '  [early fraud warning or dispute on this charge]';
    }
    console.warn(`${state.padEnd(12)} ${ch.id} ${detail}${marker}`);
  }

  console.warn(`  ${hits} of ${leaking.length} leaked charge(s) already carry fraud evidence`);
  console.warn("  guard every allow rule in Dashboard, Radar, Rules by appending " +
               "and :risk_level: != 'highest' to its predicate");
  console.warn("  then confirm the built-in rule if :risk_level: = 'highest' is still enabled");
  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_highest_risk_succeeded import verdict


def test_normal_risk_is_out_of_scope():
    assert verdict("normal", "succeeded", True, None)[0] == "baseline"


def test_unscored_charges_are_called_out_before_anything_else():
    # A charge Radar never scored cannot be blocked by any rule, so tuning
    # rules on an account full of these is wasted work.
    assert verdict("not_assessed", "succeeded", True, None)[0] == "not_assessed"
    assert verdict(None, "succeeded", True, None)[0] == "not_assessed"


def test_highest_risk_that_did_not_succeed_is_the_block_working():
    state, detail = verdict("highest", "failed", False, None)
    assert state == "stopped"
    assert "the block held" in detail


def test_an_allow_rule_is_named_when_it_overrode_the_default():
    rule = {"id": "rule_123", "action": "allow", "predicate": ":ip_country: = 'GB'"}
    state, detail = verdict("highest", "succeeded", True, rule)
    assert state == "allowed"
    assert ":ip_country: = 'GB'" in detail


def test_captured_with_no_rule_means_the_default_is_off():
    assert verdict("highest", "succeeded", True, None)[0] == "leaked"
    # A rule id string with no action is not evidence of an allow rule.
    assert verdict("highest", "succeeded", True, "rule_123")[0] == "leaked"
    # Still holdable, so it is a different instruction to a human.
    assert verdict("highest", "succeeded", False, None)[0] == "uncaptured"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict } from './stripe-highest-risk-succeeded.mjs';

test('normal risk is out of scope', () => {
  assert.equal(verdict('normal', 'succeeded', true, null)[0], 'baseline');
});

test('unscored charges are called out before anything else', () => {
  assert.equal(verdict('not_assessed', 'succeeded', true, null)[0], 'not_assessed');
  assert.equal(verdict(null, 'succeeded', true, null)[0], 'not_assessed');
});

test('highest risk that did not succeed is the block working', () => {
  const [state, detail] = verdict('highest', 'failed', false, null);
  assert.equal(state, 'stopped');
  assert.ok(detail.includes('the block held'));
});

test('an allow rule is named when it overrode the default', () => {
  const rule = { id: 'rule_123', action: 'allow', predicate: ":ip_country: = 'GB'" };
  const [state, detail] = verdict('highest', 'succeeded', true, rule);
  assert.equal(state, 'allowed');
  assert.ok(detail.includes(":ip_country: = 'GB'"));
});

test('captured with no rule means the default is off', () => {
  assert.equal(verdict('highest', 'succeeded', true, null)[0], 'leaked');
  assert.equal(verdict('highest', 'succeeded', true, 'rule_123')[0], 'leaked');
  assert.equal(verdict('highest', 'succeeded', false, null)[0], 'uncaptured');
});