From b117c2ab99a0c65b1d6218896e0589f6b28e66f3 Mon Sep 17 00:00:00 2001 From: Perseus Date: Sun, 5 Jul 2026 13:23:48 -0500 Subject: [PATCH] security: make money side effects atomic with their guards (ledger integrity) Closes three transaction-atomicity findings from the 2026-07-05 review whose shared root cause was a money side effect committing in a different transaction than the marker guarding it: - Webhook (F2/F5/F6): claim + credit/tier/refund now commit in one db.immediate transaction (commit=False threaded through mark_stripe_event/add_ledger/ set_org_tier); a crash between claim and credit no longer loses the credit, and concurrent refund/dispute events for one charge can't double-reverse. Rollback on error replaces the manual unmark. - /v1/usage idempotency (#4): the response/status is stored inside the same db.immediate block as the debits, so a crash can't leave committed events behind a reclaimable NULL-status claim that a retry would re-record (double-debit). - Hard-stop (#14): decided in integer micro-dollars, not float USD. 263 tests pass (+5: webhook apply-failure rollback, hard-stop micro boundary, idempotent atomic-store + replay-no-double-debit). Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 19 +++ plutus_agent/billing/stripe_client.py | 37 +++--- plutus_agent/db.py | 16 ++- plutus_agent/metering.py | 6 +- plutus_agent/server/app.py | 95 ++++++++------- tests/test_txn_atomicity.py | 161 ++++++++++++++++++++++++++ 6 files changed, 267 insertions(+), 67 deletions(-) create mode 100644 tests/test_txn_atomicity.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cebe65..3098358 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,25 @@ All notable changes to Plutus are documented here. ## [Unreleased] ### Security +- **Ledger atomicity: every money side effect now commits in the same transaction + as the marker guarding it.** Two crash-window bugs are closed: + (1) *Webhook silent credit loss / double-reverse* — `mark_stripe_event` committed + the event claim in a separate transaction from the credit `add_ledger`, so a + crash between them plus Stripe's retry left the event marked "duplicate" with the + credit never applied (customer paid, no credit); concurrent refund/dispute events + for one charge could also double-reverse. `handle_webhook_event` now wraps the + claim and all side effects in one `db.immediate` (`BEGIN IMMEDIATE`) transaction + with `commit=False` threaded through `mark_stripe_event`/`add_ledger`/ + `set_org_tier`; any error rolls the whole thing back for a clean retry. + (2) *`/v1/usage` idempotency double-debit* — the batch committed, but the + idempotency-key status was flipped in a later separate commit, so a crash in + between left committed events behind a NULL-status claim that the 120s reclaim + deleted, letting a retry re-record the batch. The response is now stored inside + the same `db.immediate` block as the debits. (2026-07-05 security review) +- **Prepaid hard-stop decided in integer micro-dollars.** The `block_over_balance` + check compared float USD (`balance - cost_usd < 0`); it now uses + `get_balance_micros`/`usd_to_micros` so sub-micro float error can't let a debit + slip past zero or wrongly block one. (2026-07-05 security review) - **OIDC login-CSRF: bind the OAuth flow to the browser that started it.** The authorization `state` was held only in a process-global pool, so an attacker could complete their own Google sign-in, capture their `code`+`state`, and feed diff --git a/plutus_agent/billing/stripe_client.py b/plutus_agent/billing/stripe_client.py index d01fa53..f4dc283 100644 --- a/plutus_agent/billing/stripe_client.py +++ b/plutus_agent/billing/stripe_client.py @@ -204,15 +204,20 @@ def handle_webhook_event(conn, event: dict) -> dict: """ event_id = event.get("id", "") etype = event.get("type", "") - - # Atomically claim the event first (fix #26: prevent concurrent double-credit) - if event_id and not db.mark_stripe_event(conn, event_id, etype): - return {"status": "duplicate", "type": etype, "id": event_id} - obj = (event.get("data") or {}).get("object") or {} - result = {"status": "ignored", "type": etype, "id": event_id} - try: + # Fix F2/F5/F6: claim the event AND apply its side effects in ONE serialized + # transaction. Previously the claim committed separately from the credit, so a + # crash between them + Stripe's retry left the event marked "duplicate" with + # the credit never applied (silent credit loss); and concurrent refund/dispute + # deliveries for one charge could double-reverse. BEGIN IMMEDIATE serializes + # them, and any exception rolls the whole transaction back — including the + # claim — so the event is cleanly retried (no manual unmark needed). + with db.immediate(conn): + if event_id and not db.mark_stripe_event(conn, event_id, etype, commit=False): + return {"status": "duplicate", "type": etype, "id": event_id} + + result = {"status": "ignored", "type": etype, "id": event_id} if etype == "checkout.session.completed": result = _apply_checkout_completed(conn, obj) elif etype in ("customer.subscription.created", "customer.subscription.updated"): @@ -226,11 +231,6 @@ def handle_webhook_event(conn, event: dict) -> dict: result = _apply_dispute(conn, obj) elif etype == "invoice.payment_failed": result = _apply_payment_failed(conn, obj) - except Exception: - # Rollback the event claim so it can be retried - if event_id: - db.unmark_stripe_event(conn, event_id) - raise result.setdefault("type", etype) result.setdefault("id", event_id) @@ -269,11 +269,11 @@ def _apply_checkout_completed(conn, obj) -> dict: # PaymentIntent but not always the customer — can be mapped back to it. ref = obj.get("payment_intent") or obj.get("id") row = db.add_ledger(conn, org_id, usd, "topup", - reason="Stripe checkout", stripe_ref=ref) + reason="Stripe checkout", stripe_ref=ref, commit=False) return {"status": "credited", "org_id": org_id, "amount_usd": usd, "balance_after": float(row["balance_after"])} if kind == "subscription" or obj.get("mode") == "subscription": - db.set_org_tier(conn, org_id, "pro") + db.set_org_tier(conn, org_id, "pro", commit=False) return {"status": "subscribed", "org_id": org_id, "tier": "pro"} return {"status": "ignored", "org_id": org_id} @@ -288,7 +288,7 @@ def _apply_subscription_change(conn, obj) -> dict: # restores `active` (→ Pro again) when payment succeeds, and emits # `subscription.deleted` (→ free) when dunning is exhausted. tier = "pro" if status in ("active", "trialing") else "free" - db.set_org_tier(conn, org_id, tier) + db.set_org_tier(conn, org_id, tier, commit=False) return {"status": "tier_set", "org_id": org_id, "tier": tier, "subscription_status": status} @@ -297,7 +297,7 @@ def _apply_subscription_deleted(conn, obj) -> dict: org_id = _org_from_metadata_or_customer(conn, obj) if not org_id: return {"status": "no_org"} - db.set_org_tier(conn, org_id, "free") + db.set_org_tier(conn, org_id, "free", commit=False) return {"status": "downgraded", "org_id": org_id, "tier": "free"} @@ -326,7 +326,8 @@ def _reverse(conn, org_id: str, ref: str, target_reversed_usd: float, delta = round(max(0.0, target_reversed_usd) - already, 6) if delta <= 0: return None - row = db.add_ledger(conn, org_id, -delta, kind, reason=reason, stripe_ref=ref) + row = db.add_ledger(conn, org_id, -delta, kind, reason=reason, stripe_ref=ref, + commit=False) return row, delta @@ -376,5 +377,5 @@ def _apply_payment_failed(conn, obj) -> dict: if not org_id: return {"status": "no_org"} db.log_alert(conn, org_id, "payment_failed", - "Stripe reported a failed invoice payment.") + "Stripe reported a failed invoice payment.", commit=False) return {"status": "payment_failed", "org_id": org_id} diff --git a/plutus_agent/db.py b/plutus_agent/db.py index 0f02b75..9f27822 100644 --- a/plutus_agent/db.py +++ b/plutus_agent/db.py @@ -387,9 +387,10 @@ def count_orgs_created_since(conn, since_ts: float) -> int: return int(row["n"]) -def set_org_tier(conn, org_id: str, tier: str) -> None: +def set_org_tier(conn, org_id: str, tier: str, commit: bool = True) -> None: conn.execute("UPDATE organizations SET tier=? WHERE id=?", (tier, org_id)) - conn.commit() + if commit: + conn.commit() def set_org_allow_negative(conn, org_id: str, allow: bool) -> None: @@ -774,13 +775,18 @@ def stripe_event_seen(conn, event_id: str) -> bool: ).fetchone() is not None -def mark_stripe_event(conn, event_id: str, type_: str) -> bool: - """Mark a Stripe event as processed atomically. Returns True if newly inserted.""" +def mark_stripe_event(conn, event_id: str, type_: str, commit: bool = True) -> bool: + """Mark a Stripe event as processed atomically. Returns True if newly inserted. + + Call with ``commit=False`` inside a ``db.immediate`` block so the claim is + atomic with the side effects it guards — a crash between the claim and the + credit must not leave a claimed-but-unapplied event (silent credit loss).""" cur = conn.execute( "INSERT OR IGNORE INTO stripe_events(event_id,type,processed_at) VALUES(?,?,?)", (event_id, type_, time.time()), ) - conn.commit() + if commit: + conn.commit() return cur.rowcount > 0 diff --git a/plutus_agent/metering.py b/plutus_agent/metering.py index b3c6e9e..f87e0dc 100644 --- a/plutus_agent/metering.py +++ b/plutus_agent/metering.py @@ -164,7 +164,11 @@ def record_usage(conn, org_id: str, provider: str, "SELECT 1 FROM credit_ledger WHERE org_id=? AND kind IN ('topup','grant') LIMIT 1", (org_id,), ).fetchone() - if had_credit and balance - cost_usd < 0: + # Fix (#14): decide the hard-stop in integer micro-dollars, not float USD. + # `balance` above stays float only for the balance_after display field; + # a sub-micro float error must not let a debit slip past zero (or wrongly + # block one). get_balance_micros/usd_to_micros are exact integers. + if had_credit and db.get_balance_micros(conn, org_id) - db.usd_to_micros(cost_usd) < 0: return MeterResult( event_id="", org_id=org_id, workspace_id=workspace_id, provider=provider, model=model, task_type=task_type, diff --git a/plutus_agent/server/app.py b/plutus_agent/server/app.py index 46ae7df..4426f99 100644 --- a/plutus_agent/server/app.py +++ b/plutus_agent/server/app.py @@ -535,6 +535,45 @@ def _usage_export(self, conn, path, q): return self._send(200, csv_text, "text/csv; charset=utf-8", {"Content-Disposition": 'attachment; filename="usage.csv"'}) + def _usage_response(self, conn, org_id, out, n_blocked, n_over_balance, cfg): + """Build the (code, body) for a recorded /v1/usage batch. Called inside the + db.immediate block so the stored idempotent response commits atomically + with the debits it describes.""" + from .. import metering + st = metering.tier_status(conn, org_id) + body = { + "org_id": org_id, + "tracked_tokens_mtd": st["tracked_tokens"], + "tracked_limit": st["tracked_limit"], + "tier": st["tier"], + } + n_recorded = len(out) - n_blocked - n_over_balance + if len(out) == 1: + body.update(out[0]) + else: + body["results"] = out + body["recorded"] = n_recorded + # Fix #62: surface BOTH rejection reasons. `blocked` is the *total* + # so a 200 is never mistaken for "all recorded"; the breakdown + # distinguishes free-tier quota from the prepaid hard-stop, which a + # reconciler / SDK needs to act on a partially-rejected batch. + body["blocked"] = n_blocked + n_over_balance + body["free_limit_blocked"] = n_blocked + body["over_balance_blocked"] = n_over_balance + # 402 whenever NOTHING landed, regardless of the mix of reasons (Fix #62). + if n_recorded == 0: + base = (cfg.get("auth", {}).get("base_url") or "").rstrip("/") + if n_over_balance and not n_blocked: + body["error"] = "prepaid credit exhausted" + elif n_blocked and not n_over_balance: + body["error"] = "free-tier token quota reached — upgrade to Pro" + else: + body["error"] = ("usage rejected: free-tier quota and prepaid " + "credit both exhausted") + body["upgrade_url"] = (base + "/pricing") if base else "/pricing" + return 402, body + return 200, body + def _ingest_usage(self, conn): """POST /v1/usage — meter one event (or a JSON array of them) via API key. @@ -603,8 +642,14 @@ def _ingest_usage(self, conn): replay = False try: with db.immediate(conn): - # Fix #65: claim the Idempotency-Key atomically with the recording - # so a retried/duplicated POST can't double-count or double-debit. + # Fix #65 + idempotency-reclaim double-debit fix: claim the + # Idempotency-Key AND store the response in the SAME transaction as + # the debits. Previously the claim+events committed here but the + # response/status was stored in a later separate commit — so a + # crash in between left committed events behind a NULL-status claim + # that the 120s reclaim would delete, letting a retry re-record the + # batch (double-debit). A truly-concurrent duplicate is serialized + # too: BEGIN IMMEDIATE makes the 2nd claim wait, then it replays. if idem_key and not db.claim_idempotency_key( conn, org_id, idem_key, commit=False): replay = True @@ -641,6 +686,11 @@ def _ingest_usage(self, conn): "over_balance": res.over_balance, "unpriced": res.unpriced, }) + code, body = self._usage_response( + conn, org_id, out, n_blocked, n_over_balance, cfg) + if idem_key: + db.store_idempotency_response( + conn, org_id, idem_key, code, json.dumps(body), commit=False) except Exception: return self._json(400, {"error": "batch recording failed"}) @@ -657,47 +707,6 @@ def _ingest_usage(self, conn): return self._json(409, {"error": "a request with this Idempotency-Key is still in progress"}) - st = metering.tier_status(conn, org_id) - body = { - "org_id": org_id, - "tracked_tokens_mtd": st["tracked_tokens"], - "tracked_limit": st["tracked_limit"], - "tier": st["tier"], - } - n_recorded = len(out) - n_blocked - n_over_balance - if len(out) == 1: - body.update(out[0]) - else: - body["results"] = out - body["recorded"] = n_recorded - # Fix #62: surface BOTH rejection reasons. `blocked` is the *total* - # so a 200 is never mistaken for "all recorded"; the breakdown - # distinguishes free-tier quota from the prepaid hard-stop, which a - # reconciler / SDK needs to act on a partially-rejected batch. - body["blocked"] = n_blocked + n_over_balance - body["free_limit_blocked"] = n_blocked - body["over_balance_blocked"] = n_over_balance - - # 402 whenever NOTHING landed, regardless of the mix of reasons. Fix #62: - # previously this only fired when a *single* reason accounted for the - # whole batch, so a batch split across both reasons (e.g. 60 over-balance - # + 40 over-quota) slipped through as a 200 with everything rejected. - if n_recorded == 0: - base = (cfg.get("auth", {}).get("base_url") or "").rstrip("/") - if n_over_balance and not n_blocked: - body["error"] = "prepaid credit exhausted" - elif n_blocked and not n_over_balance: - body["error"] = "free-tier token quota reached — upgrade to Pro" - else: - body["error"] = ("usage rejected: free-tier quota and prepaid " - "credit both exhausted") - body["upgrade_url"] = (base + "/pricing") if base else "/pricing" - code = 402 - else: - code = 200 - # Fix #65: persist the response so a duplicate Idempotency-Key replays it. - if idem_key: - db.store_idempotency_response(conn, org_id, idem_key, code, json.dumps(body)) return self._json(code, body) # ---- admin API (token-scoped; #66) ------------------------------------- diff --git a/tests/test_txn_atomicity.py b/tests/test_txn_atomicity.py new file mode 100644 index 0000000..e587405 --- /dev/null +++ b/tests/test_txn_atomicity.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Transaction-atomicity guarantees (2026-07-05 security review): every money +side effect commits in the SAME transaction as the marker that guards it, so a +crash or exception between them rolls BOTH back.""" +import json +import os +import sys +import tempfile +import threading +import unittest +import urllib.request + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from plutus_agent import db, metering +from plutus_agent.billing import handle_webhook_event +from plutus_agent.config import DEFAULT_CONFIG +from plutus_agent.server import app + + +class TestWebhookAtomicity(unittest.TestCase): + """F2/F5/F6: the Stripe event claim and its ledger effect commit together, so + a failure between them rolls BOTH back — no claimed-but-unapplied event + (silent credit loss) and no half-applied refund.""" + + def setUp(self): + fd, self.dbpath = tempfile.mkstemp(suffix=".db"); os.close(fd) + self.conn = db.connect(self.dbpath); db.init_schema(self.conn) + self.org = db.create_org(self.conn, "Acme")["id"] + db.set_stripe_customer(self.conn, self.org, "cus_1") + db.add_ledger(self.conn, self.org, 50.0, "topup", stripe_ref="pi_seed") + + def tearDown(self): + self.conn.close() + for ext in ("", "-wal", "-shm"): + try: + os.unlink(self.dbpath + ext) + except OSError: + pass + + def _checkout(self, event_id): + return handle_webhook_event(self.conn, { + "id": event_id, "type": "checkout.session.completed", + "data": {"object": {"id": "cs_1", "mode": "payment", "customer": "cus_1", + "amount_total": 1000, "metadata": {"kind": "credit"}, + "payment_intent": "pi_new"}}}) + + def _claim_exists(self, event_id): + return self.conn.execute( + "SELECT 1 FROM stripe_events WHERE event_id=?", (event_id,)).fetchone() is not None + + def test_apply_failure_rolls_back_claim_and_balance(self): + def boom(*a, **k): + raise RuntimeError("simulated crash mid-apply") + orig = db.add_ledger + db.add_ledger = boom + try: + with self.assertRaises(RuntimeError): + self._checkout("evt_fail") + finally: + db.add_ledger = orig + # The claim was rolled back (so Stripe's retry is honored) and no credit + # was applied — the pre-fix bug left the claim committed + credit lost. + self.assertFalse(self._claim_exists("evt_fail")) + self.assertAlmostEqual(db.get_balance(self.conn, self.org), 50.0, places=6) + # Retry now applies exactly once. + res = self._checkout("evt_fail") + self.assertEqual(res["status"], "credited") + self.assertAlmostEqual(db.get_balance(self.conn, self.org), 60.0, places=6) + + def test_duplicate_event_credits_once(self): + self.assertEqual(self._checkout("evt_dup")["status"], "credited") + self.assertEqual(self._checkout("evt_dup")["status"], "duplicate") + self.assertAlmostEqual(db.get_balance(self.conn, self.org), 60.0, places=6) + + +class TestHardStopMicros(unittest.TestCase): + """#14: the prepaid hard-stop decides in integer micro-dollars, not float USD.""" + + def setUp(self): + fd, self.dbpath = tempfile.mkstemp(suffix=".db"); os.close(fd) + self.conn = db.connect(self.dbpath); db.init_schema(self.conn) + self.org = db.create_org(self.conn, "Acme", tier="pro")["id"] + + def tearDown(self): + self.conn.close() + for ext in ("", "-wal", "-shm"): + try: + os.unlink(self.dbpath + ext) + except OSError: + pass + + def test_boundary_allows_to_exact_zero_then_blocks(self): + db.add_ledger(self.conn, self.org, 0.000002, "topup") # 2 micro-dollars + rs = [metering.record_usage(self.conn, self.org, "openai", + cost_usd=0.000001, block_over_balance=True) + for _ in range(3)] + self.assertTrue(rs[0].recorded) + self.assertTrue(rs[1].recorded) # drains to exactly 0 (0 is not < 0) + self.assertFalse(rs[2].recorded) # 0 - 1 micro < 0 -> blocked + self.assertTrue(rs[2].over_balance) + self.assertEqual(db.get_balance_micros(self.conn, self.org), 0) + + +class TestIdempotencyAtomicStore(unittest.TestCase): + """#4: the idempotent response is stored in the SAME txn as the debits, so a + successful keyed ingest never leaves a reclaimable NULL-status claim that a + later retry could re-record (double-debit).""" + + @classmethod + def setUpClass(cls): + fd, cls.dbpath = tempfile.mkstemp(suffix=".db"); os.close(fd) + conn = db.connect(cls.dbpath); db.init_schema(conn) + cls.org = db.create_org(conn, "Acme", tier="pro")["id"] + _, cls.key = db.create_api_key(conn, cls.org) + db.add_ledger(conn, cls.org, 100.0, "topup"); conn.close() + ctx = app._Ctx(dict(DEFAULT_CONFIG), cls.dbpath, demo=False) + cls.httpd = app._Server(("127.0.0.1", 0), app.Handler, ctx) + cls.port = cls.httpd.server_address[1] + threading.Thread(target=cls.httpd.serve_forever, daemon=True).start() + + @classmethod + def tearDownClass(cls): + cls.httpd.shutdown(); cls.httpd.server_close() + for ext in ("", "-wal", "-shm"): + try: + os.remove(cls.dbpath + ext) + except OSError: + pass + + def _post(self, idem): + data = json.dumps({"provider": "openai", "input_tokens": 10}).encode() + req = urllib.request.Request( + f"http://127.0.0.1:{self.port}/v1/usage", data=data, method="POST", + headers={"Authorization": f"Bearer {self.key}", + "Content-Type": "application/json", "Idempotency-Key": idem}) + r = urllib.request.urlopen(req, timeout=5) + return r.status, json.loads(r.read().decode()) + + def test_response_stored_with_non_null_status(self): + st, _ = self._post("key-atomic-1") + self.assertEqual(st, 200) + conn = db.connect(self.dbpath) + try: + row = db.idempotency_response(conn, self.org, "key-atomic-1") + finally: + conn.close() + self.assertIsNotNone(row) + self.assertIsNotNone(row[0]) # status non-NULL => never reclaimed => no double-debit + + def test_retry_replays_without_double_debit(self): + self._post("key-atomic-2") + bal1 = db.get_balance(db.connect(self.dbpath), self.org) + _, body = self._post("key-atomic-2") + self.assertTrue(body.get("idempotent_replay")) + bal2 = db.get_balance(db.connect(self.dbpath), self.org) + self.assertAlmostEqual(bal1, bal2, places=6) # retry did not debit again + + +if __name__ == "__main__": + unittest.main()