Add admin_required decorator

to mark certain commands as privileged

src/auth.py

@@ -0,0 +1,44 @@
+"""
+Logic for restricting the use of Slack commands to specific parties
+"""
+
+from functools import wraps
+
+from config import SLACK_APP
+
+
+async def get_user_info(user_id: str):
+    """
+    Queries Slack's API for information about a particular user.
+
+    See https://api.slack.com/methods/users.info
+    """
+    return await SLACK_APP.client.users_info(user=user_id)
+
+
+async def is_admin(user_id: str) -> bool:
+    """
+    Gets info for the Slack user executing the command and checks if
+    they're a workspace admin.
+    """
+    user_info = await get_user_info(user_id)
+
+    return user_info.get("user", {}).get("is_admin", False)
+
+
+def admin_required(command):
+    """
+    Used to decorate Slack commands to ensure the executor is an admin
+    before proceeding with allowing the command to run.
+    """
+
+    @wraps(command)
+    async def auth_wrapper(*args, **kwargs):
+        if await is_admin(kwargs["command"]["user_id"]):
+            return await command(*args, **kwargs)
+
+        return await kwargs["ack"](
+            f"You must be a workspace admin in order to run `{kwargs['command']['command']}`"
+        )
+
+    return auth_wrapper

src/bot.py

@@ -10,18 +10,12 @@
 
 import aiohttp
 import pytz
-from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
-from slack_bolt.async_app import AsyncApp
 
 import database
+from config import SLACK_APP
 from error import UnsafeMessageSpilloverError
 from message_builder import build_event_blocks, chunk_messages
-
-# configure app
-APP = AsyncApp(
-    token=os.environ.get("BOT_TOKEN"), signing_secret=os.environ.get("SIGNING_SECRET")
-)
-APP_HANDLER = AsyncSlackRequestHandler(APP)
+from auth import admin_required
 
 
 async def is_unsafe_to_spillover(
@@ -58,7 +52,7 @@ async def is_unsafe_to_spillover(
 
 async def post_new_message(slack_channel_id: str, msg_blocks: list, msg_text: str):
     """Posts a message to Slack"""
-    return await APP.client.chat_postMessage(
+    return await SLACK_APP.client.chat_postMessage(
         channel=slack_channel_id,
         blocks=msg_blocks,
         text=msg_text,
@@ -134,7 +128,7 @@ async def post_or_update_messages(week, messages):
                     )
 
                     timestamp = message_details[slack_channel_id][msg_idx]["timestamp"]
-                    slack_response = await APP.client.chat_update(
+                    slack_response = await SLACK_APP.client.chat_update(
                         ts=timestamp,
                         channel=slack_channel_id,
                         blocks=msg_blocks,
@@ -235,7 +229,8 @@ async def periodically_check_api():
         await asyncio.sleep(60 * 60)  # 60 minutes x 60 seconds
 
 
-@APP.command("/add_channel")
+@SLACK_APP.command("/add_channel")
+@admin_required
 async def add_channel(ack, say, logger, command):
     """Handle adding a slack channel to the bot"""
     del say
@@ -245,10 +240,11 @@ async def add_channel(ack, say, logger, command):
             await database.add_channel(command["channel_id"])
             await ack("Added channel to slack events bot 👍")
         except sqlite3.IntegrityError:
-            await ack("slack events bot has already been activated for this channel")
+            await ack("Slack events bot has already been activated for this channel")
 
 
-@APP.command("/remove_channel")
+@SLACK_APP.command("/remove_channel")
+@admin_required
 async def remove_channel(ack, say, logger, command):
     """Handle removing a slack channel from the bot"""
     del say
@@ -258,10 +254,10 @@ async def remove_channel(ack, say, logger, command):
             await database.remove_channel(command["channel_id"])
             await ack("Removed channel from slack events bot 👍")
         except sqlite3.IntegrityError:
-            await ack("slack events bot is not activated for this channel")
+            await ack("Slack events bot is not activated for this channel")
 
 
-@APP.command("/check_api")
+@SLACK_APP.command("/check_api")
 async def trigger_check_api(ack, say, logger, command):
     """Handle manually rechecking the api for updates"""
     del say

src/config.py

@@ -0,0 +1,17 @@
+"""
+Location for configuration settings and app-wide constants.
+"""
+
+import os
+
+from fastapi import FastAPI
+from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
+from slack_bolt.async_app import AsyncApp
+
+API = FastAPI()
+
+# configure Slack app
+SLACK_APP = AsyncApp(
+    token=os.environ.get("BOT_TOKEN"), signing_secret=os.environ.get("SIGNING_SECRET")
+)
+SLACK_APP_HANDLER = AsyncSlackRequestHandler(SLACK_APP)

src/server.py

@@ -14,14 +14,13 @@
 from typing import Union
 
 import uvicorn
-from fastapi import FastAPI, HTTPException, Request
+from fastapi import HTTPException, Request
 from fastapi.responses import PlainTextResponse
 from starlette.types import Message
 
 import database
-from bot import APP_HANDLER, periodically_check_api, periodically_delete_old_messages
-
-API = FastAPI()
+from bot import periodically_check_api, periodically_delete_old_messages
+from config import API, SLACK_APP_HANDLER
 
 
 async def identify_slack_team_domain(payload: bytes) -> Union[str, None]:
@@ -136,7 +135,7 @@ async def rate_limit_check_api(
 @API.post("/slack/events")
 async def slack_endpoint(req: Request):
     """The front door for all Slack requests"""
-    return await APP_HANDLER.handle(req)
+    return await SLACK_APP_HANDLER.handle(req)
 
 
 @API.get("/healthz", tags=["Utility"])