matrix-ops-bot/ops_bot/aws.py

import json
import logging
from typing import Any, List, Tuple

from fastapi import Request

from ops_bot.common import COLOR_ALARM, COLOR_OK, COLOR_UNKNOWN
from ops_bot.config import RoutingKey


def handle_subscribe_confirm(payload: Any) -> List[Tuple[str, str]]:
    message = payload.get("Message")
    url = payload.get("SubscribeURL")
    plain = f"{message}\n\n{url}"
    return [(plain, plain)]


def handle_notification(payload: Any) -> List[Tuple[str, str]]:
    message = payload.get("Message")
    subject = payload.get("Subject")

    plain = f"{subject}\n{message}"
    formatted = (
        f"<strong><font color={COLOR_ALARM}>{subject}</font></strong>\n<p>{message}</p>"
    )
    return [(plain, formatted)]


def handle_json_notification(payload: Any, body: Any) -> List[Tuple[str, str]]:
    if "AlarmName" not in body:
        msg = "Received unknown json payload type over AWS SNS"
        logging.info(msg)
        logging.info(payload.get("Message"))
        return [(msg, msg)]

    description = body.get("AlarmDescription")
    subject = payload.get("Subject")
    state_value = body.get("NewStateValue", "unknown")

    if state_value == "ALARM":
        color = COLOR_ALARM
    elif state_value == "OK":
        color = COLOR_OK
    else:
        color = COLOR_UNKNOWN

    plain = f"{subject}"
    formatted = f"<strong><font color={color}>{subject}</font></strong>\n"

    if state_value == "OK":
        plain += "\n(this alarm has been resolved!)"
        formatted += "\n<p>(this alarm has been resolved!)</p>"
    else:
        plain += "\n{description}"
        formatted += f"\n<p>{description}</p>"
    return [(plain, formatted)]


async def parse_sns_event(
    route: RoutingKey,
    payload: Any,
    request: Request,
) -> List[Tuple[str, str]]:
    if payload.get("Type") == "SubscriptionConfirmation":
        return handle_subscribe_confirm(payload)
    elif payload.get("Type") == "UnsubscribeConfirmation":
        return handle_subscribe_confirm(payload)
    elif payload.get("Type") == "Notification":
        try:
            body = json.loads(payload.get("Message"))
            return handle_json_notification(payload, body)
        except Exception:
            return handle_notification(payload)
    raise Exception("Unnown SNS payload type")