First pass at implemenation with pagerduty webhooks

ops_bot/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.1.0"

ops_bot/main.py

@@ -0,0 +1,103 @@
+import asyncio
+from typing import Any, Dict, Optional, cast
+
+import uvicorn
+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.matrix import MatrixClient, MatrixClientSettings
+
+
+class BotSettings(BaseSettings):
+    bearer_token: str
+    routing_keys: Dict[str, str]
+
+    class Config:
+        env_prefix = "BOT_"
+        case_sensitive = False
+
+
+app = FastAPI()
+security = HTTPBearer()
+
+
+async def get_matrix_service(request: Request) -> MatrixClient:
+    """A helper to fetch the matrix client from the app state"""
+    return cast(MatrixClient, request.app.state.matrix_client)
+
+
+async def matrix_main(matrix_client: MatrixClient) -> None:
+    """Execs the matrix client asyncio task"""
+    workers = [asyncio.create_task(matrix_client.start())]
+    await asyncio.gather(*workers)
+
+
+@app.on_event("startup")
+async def startup_event() -> None:
+    bot_settings = BotSettings(_env_file=".env", _env_file_encoding="utf-8")
+    matrix_settings = MatrixClientSettings(_env_file=".env", _env_file_encoding="utf-8")
+    matrix_settings.join_rooms = list(bot_settings.routing_keys.values())
+    c = MatrixClient(settings=matrix_settings)
+    app.state.matrix_client = c
+    app.state.bot_settings = bot_settings
+    asyncio.create_task(matrix_main(c))
+
+
+@app.on_event("shutdown")
+async def shutdown_event() -> None:
+    await app.state.matrix_client.shutdown()
+
+
+@app.get("/")
+async def root() -> Dict[str, str]:
+    return {"message": "Hello World"}
+
+
+def authorize(
+    request: Request, credentials: HTTPAuthorizationCredentials = Depends(security)
+) -> bool:
+    bearer_token = request.app.state.bot_settings.bearer_token
+    if credentials.credentials != bearer_token:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect bearer token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    return True
+
+
+def get_destination(bot_settings: BotSettings, routing_key: str) -> Optional[str]:
+    return bot_settings.routing_keys.get(routing_key, None)
+
+
+@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"
+        )
+    msg_plain, msg_formatted = pagerduty.parse_pagerduty_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)
+
+
+def start() -> None:
+    uvicorn.run("ops_bot.main:app", port=1111, host="0.0.0.0")  # nosec B104

ops_bot/matrix.py

@@ -0,0 +1,166 @@
+import json
+import logging
+import pathlib
+import sys
+from typing import List, Optional, Protocol
+
+from markdown import markdown
+from nio import AsyncClient, AsyncClientConfig, LoginResponse
+from pydantic import BaseModel, BaseSettings
+
+
+class ClientCredentials(BaseModel):
+    homeserver: str
+    user_id: str
+    device_id: str
+    access_token: str
+
+
+class CredentialStorage(Protocol):
+    def save_config(self, config: ClientCredentials) -> None:
+        """Save config"""
+
+    def read_config(self) -> ClientCredentials:
+        """Load config"""
+
+
+class LocalCredentialStore:
+    def __init__(self, config_file_path: pathlib.Path):
+        self.credential_file: pathlib.Path = config_file_path
+
+    def save(self, config: ClientCredentials) -> None:
+        with self.credential_file.open(mode="w") as f:
+            json.dump(config.dict(), f)
+
+    def read(self) -> ClientCredentials:
+        with self.credential_file.open(mode="r") as f:
+            return ClientCredentials(**json.load(f))
+
+    def exists(self) -> bool:
+        return self.credential_file.exists()
+
+
+class MatrixClientSettings(BaseSettings):
+    homeserver: str
+    user_id: str
+    password: str
+    device_name: str
+    store_path: str
+    join_rooms: Optional[List[str]]
+    verify_ssl: Optional[bool] = True
+
+    class Config:
+        env_prefix = "MATRIX_"
+        case_sensitive = False
+
+
+class MatrixClient:
+    def __init__(self, settings: MatrixClientSettings):
+        self.settings = settings
+        self.store_path = pathlib.Path(settings.store_path)
+        self.credential_store = LocalCredentialStore(
+            self.store_path.joinpath("credentials.json")
+        )
+
+        self.client: AsyncClient = None
+        self.client_config = AsyncClientConfig(
+            max_limit_exceeded=0,
+            max_timeouts=0,
+            store_sync_tokens=True,
+            encryption_enabled=True,
+        )
+
+        self.greeting_sent = False
+
+    async def start(self) -> None:
+        await self.login()
+
+        if self.client.should_upload_keys:
+            await self.client.keys_upload()
+
+        if self.settings.join_rooms:
+            for room in self.settings.join_rooms:
+                await self.client.join(room)
+
+        await self.client.joined_rooms()
+        await self.client.sync_forever(timeout=300000, full_state=True)
+
+    def save_credentials(self, resp: LoginResponse, homeserver: str) -> None:
+        credentials = ClientCredentials(
+            homeserver=homeserver,
+            user_id=resp.user_id,
+            device_id=resp.device_id,
+            access_token=resp.access_token,
+        )
+        self.credential_store.save(credentials)
+
+    async def login_fresh(self) -> None:
+        self.client = AsyncClient(
+            self.settings.homeserver,
+            self.settings.user_id,
+            self.settings.store_path,
+            config=self.client_config,
+            ssl=self.settings.verify_ssl,
+        )
+
+        response = await self.client.login(
+            password=self.settings.password, device_name=self.settings.device_name
+        )
+
+        if isinstance(response, LoginResponse):
+            self.save_credentials(response, self.settings.homeserver)
+        else:
+            logging.error(
+                f'Login for "{self.settings.user_id}" via homeserver="{self.settings.homeserver}"'
+            )
+            logging.info(f"Login failure response: {response}")
+            sys.exit(1)
+
+    async def login_with_credentials(self) -> None:
+        credentials = self.credential_store.read()
+        self.client = AsyncClient(
+            credentials.homeserver,
+            credentials.user_id,
+            device_id=credentials.device_id,
+            store_path=self.store_path,
+            config=self.client_config,
+            ssl=True,
+        )
+
+        self.client.restore_login(
+            user_id=credentials.user_id,
+            device_id=credentials.device_id,
+            access_token=credentials.access_token,
+        )
+
+    async def login(self) -> None:
+        if self.credential_store.exists():
+            await self.login_with_credentials()
+        else:
+            await self.login_fresh()
+
+    async def room_send(
+        self,
+        room: str,
+        message: str,
+        message_formatted: Optional[str] = None,
+    ) -> None:
+        content = {
+            "msgtype": "m.text",
+            "body": f"{message}",
+        }
+        if message_formatted is not None:
+            content["format"] = "org.matrix.custom.html"
+            content["formatted_body"] = markdown(
+                message_formatted, extensions=["extra"]
+            )
+
+        await self.client.room_send(
+            room_id=room,
+            message_type="m.room.message",
+            content=content,
+            ignore_unverified_devices=True,
+        )
+
+    async def shutdown(self) -> None:
+        await self.client.close()

ops_bot/pagerduty.py

@@ -0,0 +1,41 @@
+import json
+from typing import Any, Tuple
+
+
+def urgency_color(urgency: str) -> str:
+    if urgency == "high":
+        return "#dc3545"
+    else:
+        return "#17a2b8"
+
+
+def parse_pagerduty_event(payload: Any) -> Tuple[str, str]:
+    """
+    Parses a pagerduty webhook v3 event into a human readable message.
+    Returns a tuple where the first item is plain text, and the second item is matrix html formatted text
+    """
+    event = payload["event"]
+    # evt_id = event["id"]
+    # event_type = event["event_type"]
+    # resource_type = event["resource_type"]
+    # occurred_at = event["occurred_at"]
+    data = event["data"]
+    data_type = data["type"]
+
+    if data_type == "incident":
+        url = data["html_url"]
+        status: str = data["status"]
+        title: str = data["title"]
+        service_name: str = data["service"]["summary"]
+        urgency: str = data.get("urgency", "high")
+        plain = f"{status}: on {service_name}:  {title} {url}"
+        formatted = f"<strong><font color={urgency_color(urgency)}>{status.upper()}</font></strong> on {service_name}:  [{title}]({url})"
+        return plain, formatted
+
+    payload_str = json.dumps(payload, sort_keys=True, indent=2)
+    return (
+        "unhandled",
+        f"""**unhandled pager duty event**
+    <pre><code class="language-json">{payload_str}</code></pre>
+    """,
+    )