Merge pull request #108 from ComputerScienceHouse/slack-verification

feat: Implements Slack Signing

Author: WeatherGod3218

.env.template

@@ -7,6 +7,7 @@ CALENDAR_TIMEZONE=
 WATCHED_CHANNELS=
 SLACK_API_TOKEN=
 SLACK_JUMPSTART_MESSAGE=
+SLACK_SIGNING_SECRET=
 
 WIKI_API=
 WIKIBOT_USER=

docker-compose.yml

@@ -15,7 +15,8 @@ services:
       - WATCHED_CHANNELS=${WATCHED_CHANNELS}
       - SLACK_API_TOKEN=${SLACK_API_TOKEN}
       - SLACK_JUMPSTART_MESSAGE=${SLACK_JUMPSTART_MESSAGE}
-
+      - SLACK_SIGNING_SECRET=${SLACK_SIGNING_SECRET}
+      
       - WIKI_API=${WIKI_API}
       - WIKIBOT_USER=${WIKIBOT_USER}
       - WIKIBOT_PASSWORD=${WIKIBOT_PASSWORD}

src/api/endpoints.py

@@ -4,6 +4,7 @@
 from fastapi.responses import JSONResponse
 
 from core import slack, wikithoughts, cshcalendar
+import json
 
 logger: Logger = getLogger(__name__)
 router: APIRouter = APIRouter()
@@ -56,11 +57,25 @@ async def slack_events(request: Request) -> JSONResponse:
 		JSONResponse: A JSON response indicating the result of the event handling.
 	"""
 
-	return JSONResponse(await slack.process_slack_events(request))
+	raw_body: bytes = await request.body()
+
+	if not (slack.is_valid_slack_request(request, raw_body)):
+		logger.warning(f"Received a Fake Slack Event!: {body}")
+		return JSONResponse({"error": "Invalid signature"}, status_code=403)
+
+	body: dict = json.load(raw_body)
+
+	# Challenge from Bot Authentication
+	if request.headers.get("content-type") == "application/json":
+		if body.get("type") == "url_verification":
+			logger.info("SLACK EVENT: Was a challenge!")
+			return {"challenge": body.get("challenge")}
+
+	return JSONResponse(await slack.process_slack_events(body))
 
 
 @router.post("/slack/message_actions")
-async def message_actions(payload: str = Form(...)) -> JSONResponse:
+async def message_actions(request: Request, payload: str = Form(...)) -> JSONResponse:
 	"""
 	Handles slack message action.
 
@@ -71,6 +86,12 @@ async def message_actions(payload: str = Form(...)) -> JSONResponse:
 		JSONResponse: A JSON response indicating the result of the action.
 	"""
 
+	raw_body: bytes = await request.body()
+
+	if not (slack.is_valid_slack_request(request, raw_body)):
+		logger.warning(f"Received a Fake Slack Message Action!")
+		return JSONResponse({"error": "Invalid signature"}, status_code=403)
+
 	response_dict, status_code = await slack.process_slack_message_actions(payload)
 	return JSONResponse(response_dict, status_code=status_code)
 

src/config.py

@@ -39,6 +39,8 @@ def _get_env_variable(name: str, default: str | None = None) -> str | None:
 
 SLACK_API_TOKEN: str | None = _get_env_variable("SLACK_API_TOKEN", None)
 SLACK_JUMPSTART_MESSAGE: str = "Would you like to post this message to Jumpstart?"
+SLACK_SIGNING_SECRET: str = _get_env_variable("SLACK_SIGNING_SECRET", None)
+
 WATCHED_CHANNELS: tuple[str] = tuple(
 	_get_env_variable("WATCHED_CHANNELS", "").split(",")
 )

src/core/slack.py

@@ -7,6 +7,7 @@
 from slack_sdk.web.async_client import AsyncWebClient
 from slack_sdk.web.slack_response import SlackResponse
 from slack_sdk.errors import SlackApiError
+from slack_sdk.signature import SignatureVerifier
 
 from modules import taskmanager
 
@@ -16,6 +17,7 @@
 	SLACK_DM_TEMPLATE,
 	CALENDAR_TIMEZONE,
 	WATCHED_CHANNELS,
+	SLACK_SIGNING_SECRET,
 )
 
 from datetime import datetime
@@ -38,11 +40,6 @@
 	"RAHHHHHHHHHHHHHHHHHHHHHHHH HOW DARE YOU :skeleton-shield-banging-here:"
 )
 
-try:
-	client = AsyncWebClient(token=SLACK_API_TOKEN)
-except Exception as e:
-	logger.error(f"Failed to initialize Slack client: {e}")
-
 current_announcement: dict[str, str] = {
 	"content": "Welcome to Jumpstart!",
 	"user": "Jumpstart",
@@ -52,6 +49,41 @@
 }
 
 
+_slack_signature_verifier: SignatureVerifier | None = (
+	SignatureVerifier(SLACK_SIGNING_SECRET) if SLACK_SIGNING_SECRET else None
+)
+
+try:
+	client = AsyncWebClient(token=SLACK_API_TOKEN)
+except Exception as e:
+	logger.error(f"Failed to initialize Slack client: {e}")
+
+
+def is_valid_slack_request(request: Request, raw_body: bytes) -> bool:
+	"""
+	Validates Slack's request signature using the signing secret.
+
+	Args:
+	    request (Request): The request to be checked
+		raw_body (bytes): The raw body
+
+	Returns
+	    bool: Whether or not it's verified
+	"""
+
+	if _slack_signature_verifier is None:
+		logger.error("Slack signing secret is not configured")
+		return False
+
+	try:
+		return _slack_signature_verifier.is_valid_request(
+			body=raw_body,
+			headers=dict(request.headers),
+		)
+	except TypeError, ValueError:
+		return False
+
+
 def clean_text(raw: str) -> str:
 	"""
 	Strip Slack mrkdwn, HTML entities, and formatting characters.
@@ -186,7 +218,7 @@ async def request_upload_via_dm(user_id: str, announcement_text: str) -> None:
 		logger.error(f"Error messaging user {user_id}: {e}")
 
 
-async def process_slack_events(request: Request) -> dict[str, str]:
+async def process_slack_events(body: dict) -> dict[str, str]:
 	"""
 	Processes a slack event, logging and returning the result from the event
 
@@ -198,7 +230,6 @@ async def process_slack_events(request: Request) -> dict[str, str]:
 	"""
 
 	try:
-		body: dict = await request.json()
 		logger.info(f"Received Slack event: {body}")
 
 		event_amounts: int = get_event_retry_amount(body.get("event_id", None))
@@ -208,12 +239,6 @@ async def process_slack_events(request: Request) -> dict[str, str]:
 			)
 			return ({"status": "success"}, 200)
 
-		# Challenge from Bot Authentication
-		if request.headers.get("content-type") == "application/json":
-			if body.get("type") == "url_verification":
-				logger.info("SLACK EVENT: Was a challenge!")
-				return {"challenge": body.get("challenge")}
-
 		event: dict = body.get("event", {})
 
 		if event.get("subtype", None) is not None:
@@ -267,7 +292,9 @@ async def process_slack_message_actions(payload: str):
 			user_id = form_json.get("user", {}).get("id")
 
 			username: str = await get_username(user_id)
-			username = username[:40] # Only get the first 40 characters so it fits on a single line
+			username = username[
+				:40
+			]  # Only get the first 40 characters so it fits on a single line
 
 			add_announcement(message_object, username)