import hashlib import hmac import json import os import time import boto3 lambda_client = boto3.client("lambda") SIGNING_SECRET = os.environ["SLACK_SIGNING_SECRET"] PROCESSOR_FUNCTION_NAME = os.environ["PROCESSOR_FUNCTION_NAME"] def handler(event, context): body = event.get("body") or "" if event.get("isBase64Encoded"): import base64 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 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 base = f"v0:{timestamp}:{body}".encode("utf-8") expected = "v0=" + hmac.new(SIGNING_SECRET.encode("utf-8"), base, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature)