import base64 import hashlib import hmac import json import os import time import boto3 lambda_client = boto3.client("lambda") secrets_client = boto3.client("secretsmanager") SLACK_SIGNING_SECRET_ARN = os.environ["SLACK_SIGNING_SECRET_ARN"] PROCESSOR_FUNCTION_NAME = os.environ["PROCESSOR_FUNCTION_NAME"] _cached_signing_secret: str | None = None def handler(event, context): body = event.get("body") or "" if event.get("isBase64Encoded"): body = base64.b64decode(body).decode("utf-8") headers = {k.lower(): v for k, v in (event.get("headers") or {}).items()} timestamp = headers.get("x-slack-request-timestamp", "") signature = headers.get("x-slack-signature", "") if not verify_signature(body, timestamp, signature): return {"statusCode": 401, "body": "unauthorized"} payload = json.loads(body) if payload.get("type") == "url_verification": return { "statusCode": 200, "headers": {"Content-Type": "text/plain"}, "body": payload.get("challenge", ""), } if payload.get("type") == "event_callback": lambda_client.invoke( FunctionName=PROCESSOR_FUNCTION_NAME, InvocationType="Event", Payload=json.dumps(payload["event"]).encode("utf-8"), ) return {"statusCode": 200, "body": ""} def get_signing_secret() -> str: global _cached_signing_secret if _cached_signing_secret: return _cached_signing_secret result = secrets_client.get_secret_value(SecretId=SLACK_SIGNING_SECRET_ARN) _cached_signing_secret = result["SecretString"] return _cached_signing_secret def verify_signature(body: str, timestamp: str, signature: str) -> bool: if not timestamp or not signature: return False try: ts = int(timestamp) except ValueError: return False if abs(time.time() - ts) > 300: return False secret = get_signing_secret() base = f"v0:{timestamp}:{body}".encode("utf-8") expected = "v0=" + hmac.new(secret.encode("utf-8"), base, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature)