studio_integration_send (CWE-362): studio_integration_send.approve has no atomic reservation before apply() → one human approval fir...

Component: studio_integration_send.py — approve() (station v0.97).

The defect.
The Approval Airlock promises that an approval token is single-use and cannot re-fire an irreversible external send (studio_integration_send.py:164 / :317).

Token verification and token consumption are separated by the execution sink:

1. Lines 168–170 check token presence with os.path.isfile(spath).
2. Line 239 executes the irreversible action: result = integ.apply(plan, client, saga).
3. Lines 315–319 delete the staging file (os.remove(spath)) only after execution and only if receipt_persisted is true.

The HTTP server uses ThreadingMixIn (studio_server.py:7230) and the dispatch path (routes/dispatch_airlock.py:108) acquires no lock. The existing _APPROVE_LOCK is never taken on this path. Two concurrent approval requests for the same staging_id can both pass the isfile check before either reaches os.remove.

Reproduction (standalone harness).

import os, shutil, tempfile, json, time, threading

try:
    from workbench import studio_integration_send as s
    from workbench.primitives.sagalog import LocalSagaStore
except ImportError:  # Flat tarball or src/ layout
    import studio_integration_send as s
    from primitives.sagalog import LocalSagaStore

ws = tempfile.mkdtemp(prefix="rc_race_")
os.makedirs(os.path.join(ws, "slack_staging"), exist_ok=True)
os.makedirs(os.path.join(ws, "receipts", "capoff"), exist_ok=True)
LocalSagaStore(os.path.join(ws, "sagalog.db"))

staging_id = "stg_test_01"
plan = {"channel": "#announcements", "text": "Test"}
spath = os.path.join(ws, "slack_staging", f"{staging_id}.json")
with open(spath, "w") as f:
    json.dump({
        "staging_id": staging_id, "provider": "slack", "plan": plan,
        "action_class": "external_send",
        "integrity": s._integrity({"staging_id": staging_id, "provider": "slack", "plan": plan})
    }, f)

class CountingIntegration:
    verb, action_class, ready, env_vars = "post_message", "external_send", True, []
    def __init__(self):
        self.count, self.lock = 0, threading.Lock()
    def apply(self, plan, client, saga):
        time.sleep(0.05)
        with self.lock:
            self.count += 1
        return {"ts": f"170000000{self.count}.00", "ok": True}

integ = CountingIntegration()
s.R.REGISTRY["slack"] = integ
s._client_for = lambda i, p, w: (None, "live")

barrier = threading.Barrier(2)
responses = []

def worker(tid):
    barrier.wait()
    res = s.approve(
        "slack", staging_id, None, ws=ws,
        signing=type("S", (), {
            "SIG_VERIFIED": "SIG_VERIFIED",
            "verify_against_install": lambda *a: "SIG_VERIFIED",
            "sign_block": lambda *a: "sig"
        })(),
        policy_gate=lambda *a: {"decision": "allow"},
        integration_audit=lambda *a: None,
        audit_log=lambda *a: None
    )
    responses.append(res)

t1 = threading.Thread(target=worker, args=(1,))
t2 = threading.Thread(target=worker, args=(2,))
t1.start(); t2.start(); t1.join(); t2.join()

print("Apply count:", integ.count)
print("Both returned OK:", [r.get("ok") for r in responses])
shutil.rmtree(ws)

Expected raw output on vulnerable code:

Apply count: 2
Both returned OK: [True, True]

Scope (stated honestly).

• Affects irreversible integrations (action_class="external_send") that do not enforce their own remote idempotency.
• Requires concurrent approval requests (double-click in UI or parallel automated callers).
• This is a state-machine serialization defect in the one-time token lifecycle, not credential theft or signature forgery.

Fix.
Atomically reserve the token before calling integ.apply() by renaming spath → spath + ".inflight" with os.rename (or by acquiring _APPROVE_LOCK around the check+execute window):

--- a/studio_integration_send.py
+++ b/studio_integration_send.py
@@ -168,6 +168,13 @@
     spath = os.path.join(ws, provider + "_staging", os.path.basename(str(staging_id)) + ".json")
     if not (str(staging_id).startswith("stg_") and os.path.isfile(spath)):
         return {"ok": False, "error": "unknown or expired staging_id"}
+    # Atomic claim: rename staging file to in-flight before executing external send
+    inflight_path = spath + ".inflight"
+    try:
+        os.rename(spath, inflight_path)
+    except OSError:
+        return {"ok": False, "error": "Approval token already claimed or executing"}
+    spath = inflight_path

Classification: CWE-362