security: make money side effects atomic with their guards (ledger integrity)

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

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}

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
 
 

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,

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) -------------------------------------