mirror of
https://github.com/Sea-Haven-Industries/expense-approval-bot.git
synced 2026-05-18 20:20:14 +00:00
98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
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", "")
|
|
|
|
print(
|
|
f"Request: body_len={len(body)}, isBase64={event.get('isBase64Encoded')}, "
|
|
f"has_signature={bool(signature)}, has_timestamp={bool(timestamp)}, "
|
|
f"timestamp={timestamp}"
|
|
)
|
|
|
|
if not verify_signature(body, timestamp, signature):
|
|
print("Signature verification FAILED — returning 401")
|
|
return {"statusCode": 401, "body": "unauthorized"}
|
|
|
|
print("Signature verified")
|
|
payload = json.loads(body)
|
|
print(f"Payload type: {payload.get('type')}")
|
|
|
|
if payload.get("type") == "url_verification":
|
|
print("URL verification handshake, echoing challenge")
|
|
return {
|
|
"statusCode": 200,
|
|
"headers": {"Content-Type": "text/plain"},
|
|
"body": payload.get("challenge", ""),
|
|
}
|
|
|
|
if payload.get("type") == "event_callback":
|
|
event_type = payload.get("event", {}).get("type")
|
|
print(f"Event callback, event.type={event_type}, invoking processor")
|
|
lambda_client.invoke(
|
|
FunctionName=PROCESSOR_FUNCTION_NAME,
|
|
InvocationType="Event",
|
|
Payload=json.dumps(payload["event"]).encode("utf-8"),
|
|
)
|
|
print("Processor invoked")
|
|
|
|
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:
|
|
print("verify_signature: missing timestamp or signature header")
|
|
return False
|
|
try:
|
|
ts = int(timestamp)
|
|
except ValueError:
|
|
print(f"verify_signature: non-integer timestamp: {timestamp!r}")
|
|
return False
|
|
age = abs(time.time() - ts)
|
|
if age > 300:
|
|
print(f"verify_signature: timestamp too old (age={age:.1f}s)")
|
|
return False
|
|
secret = get_signing_secret()
|
|
secret_len = len(secret)
|
|
base = f"v0:{timestamp}:{body}".encode("utf-8")
|
|
expected = (
|
|
"v0=" + hmac.new(secret.encode("utf-8"), base, hashlib.sha256).hexdigest()
|
|
)
|
|
ok = hmac.compare_digest(expected, signature)
|
|
if not ok:
|
|
print(
|
|
f"verify_signature: HMAC mismatch. "
|
|
f"secret_len={secret_len}, expected_prefix={expected[:15]}, "
|
|
f"received_prefix={signature[:15]}"
|
|
)
|
|
return ok
|