"""Project-managed Lambda: SNS -> Lark webhook bridge.""" import json import logging import os import urllib.request import urllib.error from typing import Optional logger = logging.getLogger() logger.setLevel(logging.INFO) _webhook_url_cache = None def handler(event, context): """Lambda entry point. Processes SNS events containing CloudWatch Alarm messages.""" webhook_url = _get_webhook_url() if not webhook_url: return for record in event.get("Records", []): sns_message = record.get("Sns", {}) raw_message = sns_message.get("Message", "") # Skip SNS subscription confirmation messages if sns_message.get("Type") == "SubscriptionConfirmation": logger.info("Received SNS subscription confirmation, skipping") continue try: alarm = json.loads(raw_message) except (json.JSONDecodeError, TypeError): logger.error("Failed to parse SNS message as JSON: %s", raw_message[:200]) continue # Verify this is a CloudWatch Alarm message if "AlarmName" not in alarm: logger.warning("Message is not a CloudWatch Alarm, skipping: %s", raw_message[:200]) continue card = _build_lark_card(alarm) _send_to_lark(webhook_url, card) def _build_lark_card(alarm: dict) -> dict: """Build a Lark interactive card from a CloudWatch Alarm message.""" alarm_name = alarm.get("AlarmName", "Unknown") new_state = alarm.get("NewStateValue", "UNKNOWN") old_state = alarm.get("OldStateValue", "UNKNOWN") reason = alarm.get("NewStateReason", "No reason provided") state_change_time = alarm.get("StateChangeTime", "Unknown") region = alarm.get("Region", "Unknown") aws_account_id = alarm.get("AWSAccountId", "Unknown") description = alarm.get("AlarmDescription", "") # Extract endpoint info from description or alarm name endpoint = description if description else alarm_name # Extract Health Check ID from Trigger dimensions health_check_id = "Unknown" trigger = alarm.get("Trigger", {}) dimensions = trigger.get("Dimensions", []) for dim in dimensions: if dim.get("name") == "HealthCheckId" or dim.get("Name") == "HealthCheckId": health_check_id = dim.get("value") or dim.get("Value", "Unknown") break # Protocol is derived from health-check metric namespace; always HTTPS/HTTP/TCP # but the alarm JSON itself doesn't carry it, so fall back to a generic label. protocol = "Route53 Health Check" if new_state == "ALARM": header_title = f"🔴 Route53 ALARM: {alarm_name}" header_template = "red" status_text = "ALARM" fields = [ _field("Endpoint", endpoint), _field("Protocol", protocol), _field("Health Check ID", health_check_id), _field("AWS Account", aws_account_id), _field("Region", region), _field("Status", status_text), _field("Time", state_change_time), ] elements = [ {"tag": "div", "fields": fields}, { "tag": "div", "text": {"tag": "lark_md", "content": f"**Failure Reason**\n{reason}"}, }, {"tag": "hr"}, { "tag": "note", "elements": [ {"tag": "plain_text", "content": f"Source: Route 53 Health Check | Region: {region}"} ], }, ] elif new_state == "OK": header_title = f"✅ Route53 OK: {alarm_name}" header_template = "green" fields = [ _field("Endpoint", endpoint), _field("AWS Account", aws_account_id), _field("Region", region), _field("Status", "Recovered"), _field("Time", state_change_time), _field("Previous", old_state), ] elements = [ {"tag": "div", "fields": fields}, {"tag": "hr"}, { "tag": "note", "elements": [ {"tag": "plain_text", "content": f"Source: Route 53 Health Check | Region: {region}"} ], }, ] else: header_title = f"🟡 Route53 INSUFFICIENT_DATA: {alarm_name}" header_template = "yellow" fields = [ _field("Endpoint", endpoint), _field("Protocol", protocol), _field("Health Check ID", health_check_id), _field("AWS Account", aws_account_id), _field("Region", region), _field("Status", new_state), _field("Previous", old_state), _field("Time", state_change_time), ] elements = [ {"tag": "div", "fields": fields}, { "tag": "div", "text": {"tag": "lark_md", "content": f"**Reason**\n{reason}"}, }, {"tag": "hr"}, { "tag": "note", "elements": [ {"tag": "plain_text", "content": f"Source: Route 53 Health Check | Region: {region}"} ], }, ] return { "msg_type": "interactive", "card": { "header": { "title": {"tag": "plain_text", "content": header_title}, "template": header_template, }, "elements": elements, }, } def _field(label: str, value: str) -> dict: """Build a Lark card field.""" return { "is_short": True, "text": {"tag": "lark_md", "content": f"**{label}**\n{value}"}, } def _send_to_lark(webhook_url: str, card: dict) -> None: """POST a card message to Lark webhook.""" payload = json.dumps(card).encode("utf-8") req = urllib.request.Request( webhook_url, data=payload, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=10) as resp: body = resp.read().decode("utf-8") logger.info("Lark webhook response: %s %s", resp.status, body) except urllib.error.HTTPError as e: body = e.read().decode("utf-8") if e.fp else "" logger.error("Lark webhook HTTP error %s: %s", e.code, body) raise except urllib.error.URLError as e: logger.error("Lark webhook connection error: %s", e.reason) raise def _get_webhook_url() -> Optional[str]: """Load the Lark webhook URL from Secrets Manager, with legacy env fallback.""" global _webhook_url_cache if _webhook_url_cache: return _webhook_url_cache secret_arn = os.environ.get("LARK_WEBHOOK_SECRET_ARN", "").strip() if secret_arn: client = _get_secrets_manager_client() response = client.get_secret_value(SecretId=secret_arn) secret_string = (response.get("SecretString") or "").strip() if not secret_string: logger.error("LARK_WEBHOOK_SECRET_ARN resolved but SecretString was empty") return None _webhook_url_cache = secret_string return _webhook_url_cache legacy_webhook_url = os.environ.get("LARK_WEBHOOK_URL", "").strip() if legacy_webhook_url: logger.warning("Using legacy LARK_WEBHOOK_URL environment variable; migrate to Secrets Manager") _webhook_url_cache = legacy_webhook_url return _webhook_url_cache logger.error("No Lark webhook configuration found in Secrets Manager or environment") return None def _get_secrets_manager_client(): """Create a boto3 Secrets Manager client lazily for local test compatibility.""" import boto3 # Imported lazily because Lambda runtime provides it but local tests may not. return boto3.client("secretsmanager")