Add SNS format

ops_bot/aws.py

@@ -0,0 +1,33 @@
+import json
+from typing import Any, Tuple
+
+from ops_bot.common import urgency_color
+
+
+def handle_subscribe_confirm(payload: Any) -> 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) -> Tuple[str, str]:
+    message = payload.get("Message")
+    subject = payload.get("Subject")
+
+    plain = f"{subject}\n{message}"
+    color = urgency_color("high")
+    formatted = (
+        f"<strong><font color={color}>{subject}</font></strong>\n<p>{message}</p>"
+    )
+    return plain, formatted
+
+
+def parse_sns_event(payload: Any) -> 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":
+        return handle_notification(payload)
+    raise Exception("Unnown SNS payload type")

ops_bot/common.py

@@ -0,0 +1,6 @@
+def urgency_color(urgency: str) -> str:
+    if urgency == "high":
+        return "#dc3545"  # red
+    else:
+        return "#ffc107"  # orange
+        # return "#17a2b8" # blue

ops_bot/main.py

@@ -8,7 +8,7 @@ from fastapi import Depends, FastAPI, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from pydantic import BaseSettings
 
-from ops_bot import pagerduty
+from ops_bot import aws, pagerduty
 from ops_bot.matrix import MatrixClient, MatrixClientSettings
 
 
@@ -77,22 +77,27 @@ def get_destination(bot_settings: BotSettings, routing_key: str) -> Optional[str
     return bot_settings.routing_keys.get(routing_key, None)
 
 
+async def receive_helper(request: Request):
+    payload: Any = await request.json()
+    routing_key = request.path_params["routing_key"]
+    room_id = get_destination(request.app.state.bot_settings, routing_key)
+    if room_id is None:
+        logging.error(f"unknown routing key {routing_key}")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Unknown routing key"
+        )
+    payload_str = json.dumps(payload, sort_keys=True, indent=2)
+    logging.info(f"received payload: \n {payload_str}")
+    return payload
+
+
 @app.post("/hook/pagerduty/{routing_key}")
 async def pagerduty_hook(
     request: Request,
     matrix_client: MatrixClient = Depends(get_matrix_service),
     auth: bool = Depends(authorize),
 ) -> Dict[str, str]:
-    payload: Any = await request.json()
-    room_id = get_destination(
-        request.app.state.bot_settings, request.path_params["routing_key"]
-    )
-    if room_id is None:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="Unknown routing key"
-        )
-    payload_str = json.dumps(payload, sort_keys=True, indent=2)
-    logging.info(f"received pagerduty payload: \n {payload_str}")
+    room_id, payload = await receive_helper(request)
     msg_plain, msg_formatted = pagerduty.parse_pagerduty_event(payload)
     await matrix_client.room_send(
         room_id,
@@ -102,6 +107,22 @@ async def pagerduty_hook(
     return {"message": msg_plain, "message_formatted": msg_formatted}
 
 
+@app.post("/hook/aws-sns/{routing_key}")
+async def aws_sns_hook(
+    request: Request,
+    matrix_client: MatrixClient = Depends(get_matrix_service),
+    auth: bool = Depends(authorize),
+) -> Dict[str, str]:
+    room_id, payload = await receive_helper(request)
+    msg_plain, msg_formatted = aws.parse_sns_event(payload)
+    await matrix_client.room_send(
+        room_id,
+        msg_plain,
+        message_formatted=msg_formatted,
+    )
+    return {"message": msg_plain, "message_formatted": msg_formatted}
+
+
 def start_dev() -> None:
     uvicorn.run("ops_bot.main:app", port=1111, host="127.0.0.1", reload=True)
 

ops_bot/pagerduty.py

@@ -1,13 +1,7 @@
 import json
 from typing import Any, Tuple
 
-
-def urgency_color(urgency: str) -> str:
-    if urgency == "high":
-        return "#dc3545"  # red
-    else:
-        return "#ffc107"  # orange
-        # return "#17a2b8" # blue
+from ops_bot.common import urgency_color
 
 
 def parse_pagerduty_event(payload: Any) -> Tuple[str, str]: