fix(security): enforce user authorization on approval button clicks
Approval button clicks (Block Kit actions in Slack, CallbackQuery in Telegram) bypass the normal message authorization flow in gateway/run.py. Any workspace/group member who can see the approval message could click Approve to authorize dangerous commands. Read SLACK_ALLOWED_USERS / TELEGRAM_ALLOWED_USERS env vars directly in the approval handlers. When an allowlist is configured and the clicking user is not in it, the click is silently ignored (Slack) or answered with an error (Telegram). Wildcard '*' permits all users. When no allowlist is configured, behavior is unchanged (open access). Based on the idea from PR #6735 by maymuneth, reimplemented to use the existing env-var-based authorization system rather than a nonexistent _allowed_user_ids adapter attribute.
This commit is contained in:
@@ -1229,6 +1229,20 @@ class SlackAdapter(BasePlatformAdapter):
|
||||
msg_ts = message.get("ts", "")
|
||||
channel_id = body.get("channel", {}).get("id", "")
|
||||
user_name = body.get("user", {}).get("name", "unknown")
|
||||
user_id = body.get("user", {}).get("id", "")
|
||||
|
||||
# Only authorized users may click approval buttons. Button clicks
|
||||
# bypass the normal message auth flow in gateway/run.py, so we must
|
||||
# check here as well.
|
||||
allowed_csv = os.getenv("SLACK_ALLOWED_USERS", "").strip()
|
||||
if allowed_csv:
|
||||
allowed_ids = {uid.strip() for uid in allowed_csv.split(",") if uid.strip()}
|
||||
if "*" not in allowed_ids and user_id not in allowed_ids:
|
||||
logger.warning(
|
||||
"[Slack] Unauthorized approval click by %s (%s) — ignoring",
|
||||
user_name, user_id,
|
||||
)
|
||||
return
|
||||
|
||||
# Map action_id to approval choice
|
||||
choice_map = {
|
||||
|
||||
@@ -1398,6 +1398,15 @@ class TelegramAdapter(BasePlatformAdapter):
|
||||
await query.answer(text="Invalid approval data.")
|
||||
return
|
||||
|
||||
# Only authorized users may click approval buttons.
|
||||
caller_id = str(getattr(query.from_user, "id", ""))
|
||||
allowed_csv = os.getenv("TELEGRAM_ALLOWED_USERS", "").strip()
|
||||
if allowed_csv:
|
||||
allowed_ids = {uid.strip() for uid in allowed_csv.split(",") if uid.strip()}
|
||||
if "*" not in allowed_ids and caller_id not in allowed_ids:
|
||||
await query.answer(text="⛔ You are not authorized to approve commands.")
|
||||
return
|
||||
|
||||
session_key = self._approval_state.pop(approval_id, None)
|
||||
if not session_key:
|
||||
await query.answer(text="This approval has already been resolved.")
|
||||
|
||||
Reference in New Issue
Block a user