Abuse case library: tests and controls by feature

# One abuse case, stored next to the story it belongs to (for example docs/abuse-cases/AC-PAY-1.yaml).
id: AC-PAY-1
feature: checkout / coupons
title: Redeem a single-use coupon more than once
attacker:
  who: Signed-in customer with one valid single-use coupon
  goal: Apply the coupon to many orders
preconditions:
  - Coupon redemption is an HTTP request the client can send in parallel
  - The server checks "unused" and marks "used" in separate steps
test:
  type: automated
  steps: Send 20 redemption requests for the same coupon at the same moment, each for a different order
  expected: Exactly one succeeds; the others get 409 and no discount
  location: tests/test_abuse_cases.py::test_ac_pay_1_single_use_coupon_under_parallel_requests
control: A single conditional UPDATE ... WHERE used_by IS NULL, checked by its affected row count
mappings:
  cwe: [CWE-362, CWE-837]
  capec: [CAPEC-26]
  asvs: [v5.0.0-2.3.3, v5.0.0-2.3.4, v5.0.0-15.4.2]
owner: payments team
status: tested

import hashlib
import hmac
import sqlite3
import time


def redeem(db_path, code, order_id):
    """AC-PAY-1 control: one conditional UPDATE, so only one request can win."""
    with sqlite3.connect(db_path, timeout=10, isolation_level=None) as db:
        won = db.execute(
            "UPDATE coupons SET used_by = ? WHERE code = ? AND used_by IS NULL",
            (order_id, code),
        ).rowcount
    return won == 1


class WebhookVerifier:
    """AC-WH-1 and AC-WH-2 controls: verify the HMAC, the timestamp and the delivery ID."""

    def __init__(self, secret, tolerance=300, clock=time.time):
        self.secret, self.tolerance, self.clock = secret, tolerance, clock
        self.seen = {}  # delivery id -> timestamp; use a shared store with a TTL in production

    def verify(self, delivery_id, timestamp, body, signature):
        expected = hmac.new(
            self.secret, f"{delivery_id}.{timestamp}.".encode() + body, hashlib.sha256
        ).hexdigest()
        if not hmac.compare_digest(expected, signature):
            return "bad-signature"
        if abs(self.clock() - int(timestamp)) > self.tolerance:
            return "stale"
        if delivery_id in self.seen:
            return "replay"
        self.seen[delivery_id] = int(timestamp)
        return "ok"

import hashlib
import hmac
import sqlite3
import threading
from concurrent.futures import ThreadPoolExecutor

import pytest

from abuse_cases import WebhookVerifier, redeem


@pytest.fixture
def coupon_db(tmp_path):
    path = tmp_path / "shop.db"
    with sqlite3.connect(path) as db:
        db.execute("CREATE TABLE coupons (code TEXT PRIMARY KEY, used_by TEXT)")
        db.execute("INSERT INTO coupons VALUES ('WELCOME10', NULL)")
    return path


def test_ac_pay_1_single_use_coupon_under_parallel_requests(coupon_db):
    # A race may not show on every run, so repeat it; any double redemption fails the test.
    for round_ in range(25):
        with sqlite3.connect(coupon_db) as db:
            db.execute("UPDATE coupons SET used_by = NULL")
        start = threading.Barrier(20)

        def attempt(n):
            start.wait()  # release all 20 requests at once
            return redeem(coupon_db, "WELCOME10", f"order-{round_}-{n}")

        with ThreadPoolExecutor(max_workers=20) as pool:
            results = list(pool.map(attempt, range(20)))
        assert results.count(True) == 1, f"round {round_}: {results.count(True)} redemptions"


SECRET = b"whsec_test_only"


def sign(delivery_id, timestamp, body):
    message = f"{delivery_id}.{timestamp}.".encode() + body
    return hmac.new(SECRET, message, hashlib.sha256).hexdigest()


@pytest.fixture
def verifier():
    return WebhookVerifier(SECRET, clock=lambda: 1_760_000_000)


def test_ac_wh_1_forged_delivery_is_rejected(verifier):
    body = b'{"order":"o-7","status":"paid"}'
    assert verifier.verify("d-1", "1760000000", body, "0" * 64) == "bad-signature"


def test_ac_wh_1_changed_body_is_rejected(verifier):
    signature = sign("d-1", "1760000000", b'{"order":"o-7","status":"failed"}')
    assert verifier.verify("d-1", "1760000000", b'{"order":"o-7","status":"paid"}', signature) == "bad-signature"


def test_ac_wh_2_replayed_delivery_is_rejected(verifier):
    body = b'{"order":"o-7","status":"paid"}'
    signature = sign("d-2", "1760000000", body)
    assert verifier.verify("d-2", "1760000000", body, signature) == "ok"
    assert verifier.verify("d-2", "1760000000", body, signature) == "replay"


def test_ac_wh_2_old_delivery_is_rejected(verifier):
    body = b'{"order":"o-7","status":"paid"}'
    signature = sign("d-3", "1759990000", body)  # 10,000 seconds old
    assert verifier.verify("d-3", "1759990000", body, signature) == "stale"