""" SNS -> Slack subscriber for the humanwork monitoring module. Receives a CloudWatch-alarm-shaped SNS message, formats it as a Slack block, posts to the webhook URL in Secrets Manager. Webhook URL is fetched once per cold start and cached for the lifetime of the execution environment. Module-level invariants: * The Slack webhook secret value is the raw URL (no JSON envelope). This matches how operators paste a Slack incoming-webhook URL into the AWS console. * SNS payload shape is the CloudWatch-alarm-on-SNS envelope: {"Records": [{"Sns": {"Message": ""}}]} so we json.loads the Message twice (envelope, then alarm event). * The handler never raises on a malformed payload — it logs and returns "ok" so SNS does not retry forever. """ from __future__ import annotations import json import logging import os import urllib.error import urllib.request import boto3 logger = logging.getLogger() logger.setLevel(logging.INFO) _SECRET_ARN = os.environ["SLACK_WEBHOOK_SECRET_ARN"] _ENVIRONMENT = os.environ.get("ENVIRONMENT", "unknown") _secrets_client = boto3.client("secretsmanager") _cached_webhook_url: str | None = None def _get_webhook_url() -> str: """Fetch the Slack webhook URL from Secrets Manager. Cached per cold start.""" global _cached_webhook_url if _cached_webhook_url is None: resp = _secrets_client.get_secret_value(SecretId=_SECRET_ARN) # SecretString is the canonical field; SecretBinary path not used here. value = resp.get("SecretString") if not value: raise RuntimeError(f"Slack webhook secret {_SECRET_ARN} has no SecretString") _cached_webhook_url = value.strip() return _cached_webhook_url # #3990 — structured response-metadata suffix on AlarmDescription: # " | severity=P1 | owner=infra | runbook=docs/runbooks/x.md" # Parsed here (SNS alarm payloads do NOT carry alarm tags, so the description # is the only self-contained channel for this metadata). Unknown keys are # ignored; a description without the suffix renders exactly as before. _STRUCTURED_KEYS = ("severity", "owner", "runbook") def _parse_description(description: str) -> tuple[str, dict]: """Split ' | key=value | ...' into (prose, {key: value}).""" if not description or " | " not in description: return description, {} parts = description.split(" | ") prose_parts = [parts[0]] meta: dict = {} for part in parts[1:]: if "=" in part: key, _, value = part.partition("=") key = key.strip().lower() if key in _STRUCTURED_KEYS and value.strip(): meta[key] = value.strip() continue prose_parts.append(part) return " | ".join(prose_parts), meta def _severity_color(new_state: str) -> str: """Slack attachment color by alarm state. ALARM=red, OK=green, INSUFFICIENT_DATA=yellow.""" if new_state == "ALARM": return "#dc2626" if new_state == "OK": return "#16a34a" return "#f59e0b" def _format_blocks(event: dict) -> dict: """Turn a CloudWatch alarm SNS event into a Slack-blocks payload.""" name = event.get("AlarmName", "unknown") raw_description = event.get("AlarmDescription", "") description, response_meta = _parse_description(raw_description) new_state = event.get("NewStateValue", "ALARM") reason = event.get("NewStateReason", "") region = event.get("Region", "unknown") timestamp = event.get("StateChangeTime", "") # Build a Slack message that survives both rich-block-aware and # plain-text clients. Fall back to the `text` field when blocks fail. title_emoji = ":red_circle:" if new_state == "ALARM" else ":large_green_circle:" if new_state == "OK" else ":large_yellow_circle:" title = f"{title_emoji} {new_state}: `{name}`" return { "text": f"{new_state}: {name} — {description}", "attachments": [ { "color": _severity_color(new_state), "blocks": [ { "type": "header", "text": {"type": "plain_text", "text": title[:150]}, }, { "type": "section", "fields": [ {"type": "mrkdwn", "text": f"*Environment*\n`{_ENVIRONMENT}`"}, {"type": "mrkdwn", "text": f"*Region*\n`{region}`"}, {"type": "mrkdwn", "text": f"*State*\n`{new_state}`"}, {"type": "mrkdwn", "text": f"*Time*\n{timestamp}"}, ] + ( # #3990 — response metadata: who owns it, how urgent, # which handbook. Rendered only when present. [ { "type": "mrkdwn", "text": f"*Severity*\n`{response_meta['severity']}`", } ] if "severity" in response_meta else [] ) + ( [ { "type": "mrkdwn", "text": f"*Owner*\n`{response_meta['owner']}`", } ] if "owner" in response_meta else [] ) + ( [ { "type": "mrkdwn", "text": f"*Runbook*\n`{response_meta['runbook']}`", } ] if "runbook" in response_meta else [] ), }, { "type": "section", "text": { "type": "mrkdwn", "text": f"*Description*\n{description or '_(no description)_'}", }, }, { "type": "section", "text": { "type": "mrkdwn", "text": f"*Reason*\n```{reason[:2500]}```", }, }, ], } ], } def _post_to_slack(payload: dict) -> None: url = _get_webhook_url() req = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=5) as resp: body = resp.read().decode("utf-8", errors="replace") logger.info("slack_post status=%s body=%s", resp.status, body[:200]) except urllib.error.HTTPError as e: logger.error("slack_post HTTPError: %s %s", e.code, e.read()[:200].decode("utf-8", errors="replace")) except Exception as e: # noqa: BLE001 logger.error("slack_post failed: %s", e) def handler(event, _context): # noqa: D401 """SNS handler. Always returns ok so SNS does not retry forever.""" try: for record in event.get("Records", []): sns = record.get("Sns", {}) raw = sns.get("Message", "") try: alarm_event = json.loads(raw) except json.JSONDecodeError: # Plain-text SNS message — wrap it minimally so the formatter # still produces something useful in Slack. alarm_event = {"AlarmName": sns.get("Subject", "sns-message"), "AlarmDescription": raw} payload = _format_blocks(alarm_event) _post_to_slack(payload) except Exception as e: # noqa: BLE001 # Never raise — SNS would retry indefinitely and spam the channel. logger.error("handler unhandled error: %s", e) return {"ok": True}