routes/dispatch_airlock.py's /api/mcp/pending/approve handler
(_handle_mcp_pending_approve) takes two request-body fields to locate a
pending staging record: file and source. file is correctly passed
through os.path.basename() before it ever touches a path, stripping any
directory components. source never gets the same treatment -- it's used
directly inside os.path.join(WS, source, ...) whenever it ends with the
literal string "_staging".
Because os.path.join() doesn't stop at a directory boundary, a source
value like "../../../somewhere_staging" resolves outside the workspace
entirely. The handler then reads whatever JSON file it finds there and,
if that file has provider and staging_id/consent_token fields
(the same shape a legitimate staging record has), passes those values
straight into studio_integration_send.approve() -- the live governed-
send approval engine, the same code path a real human clicking "Approve"
in the Sends-tab banner triggers.
Reproduction steps:
- Extract a clean station-v0.73 tarball, sys.path.insert(0, "workbench").
- Create a JSON file OUTSIDE the intended workspace directory, e.g.
/tmp/outside_staging/stg_evil.json containing
{"provider": "stripe", "staging_id": "<some token>", "verb": "..."}.
- import routes.dispatch_airlock as DA; set DA.WS to a REAL workspace
directory elsewhere (not the one from step 2).
- Call DA._handle_mcp_pending_approve(
{"source": "../../../../tmp/outside_staging", "file": "stg_evil.json"},
handler)
with a handler stub that authenticates the session (_require_session
returning True, exactly like a real authenticated Studio session
would).
Expected: the handler should refuse a source value that resolves
outside the workspace, exactly like it already refuses a file value
containing a path separator.
Actual: the handler reads the file OUTSIDE the workspace and forwards itsprovider/staging_id fields directly into the approval engine as if it
were a legitimate, locally-staged pending action.
Root cause: routes/dispatch_airlock.py, _handle_mcp_pending_approve()
(~line 106-134):
fname = str(b.get("file") or "").strip()
source = str(b.get("source") or "").strip()
...
elif source.endswith("_staging"):
fpath = os.path.join(WS, source, os.path.basename(fname))
os.path.basename(fname) sanitizes file; nothing equivalent is ever
applied to source, even though it's the field actually used as the
directory component of the path.
Suggested fix: validate source the same way file already is -- either
require it to be exactly one of the known staging directory names (a
small, enumerable set: <provider>_staging for each registered
provider), or strip it down with os.path.basename() before joining, so a
traversal sequence can never make it into the path at all.