Refactor/api webhooks handler (#949)

* fix: ensure proper handling of alert/info hook type

* feat: isolate log_ops_message feature and handle errors

* feat: add configuration settings for SRE Ops

* refactor: fix and reorder import statements

* fix: removed former log_ops_message function

* refactor: migrate webhook payload handling to dedicated module

* refactor: migrate AWS SNS handling to dedicated module

* chore: remove migrated functions modules

* refactor: isolate AWS processor to dedicated module

* refactor: isolate AWS processor to dedicated module

* feat: add SlackClientManager

* refactor: simplify log_ops_message by using SlackClientManager

* refactor: remove client param from log_ops_message calls

* chore: fmt

Author: gcharest

app/api/v1/routes/webhooks.py

@@ -1,22 +1,15 @@
 import json
-from typing import Union, Dict, Any, cast
+from typing import Union, Dict, Any
 
-import requests  # type: ignore
 from api.dependencies.rate_limits import get_limiter
 from core.logging import get_module_logger
 from fastapi import APIRouter, HTTPException, Request, Body
 from integrations.sentinel import log_to_sentinel
 from models.webhooks import (
-    AwsSnsPayload,
     WebhookPayload,
-    AccessRequest,
-    UpptimePayload,
-    WebhookResult,
 )
 from modules.slack import webhooks
-from modules.webhooks.base import validate_payload
-from server.event_handlers import aws
-from server.utils import log_ops_message
+from modules.webhooks.base import handle_webhook_payload
 
 
 logger = get_module_logger()
@@ -107,125 +100,6 @@ def handle_webhook(
     return {"ok": True}
 
 
-def handle_webhook_payload(
-    payload_dict: dict,
-    request: Request,
-) -> WebhookResult:
-    """Process and validate the webhook payload.
-
-    Returns:
-        dict: A dictionary containing:
-            - status (str): The status of the operation (e.g., "success", "error").
-            - action (Literal["post", "log", "none"]): The action to take.
-            - payload (Optional[WebhookPayload]): The payload to post, if applicable.
-    """
-    logger.info("processing_webhook_payload", payload=payload_dict)
-    payload_validation_result = validate_payload(payload_dict)
-
-    webhook_result = WebhookResult(
-        status="error", message="Failed to process payload for unknown reasons"
-    )
-    if payload_validation_result is not None:
-        payload_type, validated_payload = payload_validation_result
-    else:
-        error_message = "No matching model found for payload"
-        return WebhookResult(status="error", message=error_message)
-
-    match payload_type.__name__:
-        case "WebhookPayload":
-            webhook_result = WebhookResult(
-                status="success", action="post", payload=validated_payload
-            )
-        case "AwsSnsPayload":
-            aws_sns_payload_instance = cast(AwsSnsPayload, validated_payload)
-            aws_sns_payload = aws.validate_sns_payload(
-                aws_sns_payload_instance,
-                request.state.bot.client,
-            )
-
-            if aws_sns_payload.Type == "SubscriptionConfirmation":
-                requests.get(aws_sns_payload.SubscribeURL, timeout=60)
-                logger.info(
-                    "subscribed_webhook_to_topic",
-                    webhook_id=aws_sns_payload.TopicArn,
-                    subscribed_topic=aws_sns_payload.TopicArn,
-                )
-                log_ops_message(
-                    request.state.bot.client,
-                    f"Subscribed webhook {id} to topic {aws_sns_payload.TopicArn}",
-                )
-                webhook_result = WebhookResult(
-                    status="success", action="log", payload=None
-                )
-
-            if aws_sns_payload.Type == "UnsubscribeConfirmation":
-                log_ops_message(
-                    request.state.bot.client,
-                    f"{aws_sns_payload.TopicArn} unsubscribed from webhook {id}",
-                )
-                webhook_result = WebhookResult(
-                    status="success", action="log", payload=None
-                )
-
-            if aws_sns_payload.Type == "Notification":
-                blocks = aws.parse(aws_sns_payload, request.state.bot.client)
-                if not blocks:
-                    logger.info(
-                        "payload_empty_message",
-                        payload_type="AwsSnsPayload",
-                        sns_type=aws_sns_payload.Type,
-                    )
-                    return WebhookResult(
-                        status="error",
-                        action="none",
-                        message="Empty AWS SNS Notification message",
-                    )
-                webhook_result = WebhookResult(
-                    status="success",
-                    action="post",
-                    payload=WebhookPayload(blocks=blocks),
-                )
-
-        case "AccessRequest":
-            message = str(cast(AccessRequest, validated_payload).model_dump())
-            webhook_result = WebhookResult(
-                status="success",
-                action="post",
-                payload=WebhookPayload(text=message),
-            )
-
-        case "UpptimePayload":
-            text = cast(UpptimePayload, validated_payload).text
-            header_text = "📈 Web Application Status Changed!"
-            blocks = [
-                {"type": "section", "text": {"type": "mrkdwn", "text": " "}},
-                {
-                    "type": "header",
-                    "text": {"type": "plain_text", "text": f"{header_text}"},
-                },
-                {
-                    "type": "section",
-                    "text": {
-                        "type": "mrkdwn",
-                        "text": f"{text}",
-                    },
-                },
-            ]
-            webhook_result = WebhookResult(
-                status="success",
-                action="post",
-                payload=WebhookPayload(blocks=blocks),
-            )
-
-        case _:
-            webhook_result = WebhookResult(
-                status="error",
-                message="No matching model found for payload",
-            )
-
-    return webhook_result
-
-
 def append_incident_buttons(payload: WebhookPayload, webhook_id) -> WebhookPayload:
     if payload.attachments is None:
         payload.attachments = []

app/core/config.py

@@ -190,6 +190,8 @@ class TalentRoleSettings(BaseSettings):
 
 
 class ReportsSettings(BaseSettings):
+    """Reports configuration settings."""
+
     FOLDER_REPORTS_GOOGLE_GROUPS: str = Field(
         default="", alias="FOLDER_REPORTS_GOOGLE_GROUPS"
     )
@@ -202,6 +204,8 @@ class ReportsSettings(BaseSettings):
 
 
 class AWSFeatureSettings(BaseSettings):
+    """AWS Feature configuration settings."""
+
     AWS_ADMIN_GROUPS: list[str] = Field(
         default=["[email protected]"], alias="AWS_ADMIN_GROUPS"
     )
@@ -234,6 +238,17 @@ class IncidentFeatureSettings(BaseSettings):
     )
 
 
+class SreOpsSettings(BaseSettings):
+    """SRE Ops configuration settings."""
+
+    SRE_OPS_CHANNEL_ID: str = Field(default="", alias="SRE_OPS_CHANNEL_ID")
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        case_sensitive=True,
+        extra="ignore",
+    )
+
+
 class ServerSettings(BaseSettings):
     """Server configuration settings."""
 
@@ -319,6 +334,7 @@ class Settings(BaseSettings):
     reports: ReportsSettings
     aws_feature: AWSFeatureSettings
     feat_incident: IncidentFeatureSettings
+    sre_ops: SreOpsSettings
 
     # Development settings
     dev: DevSettings
@@ -345,6 +361,7 @@ def __init__(self, **kwargs):
             "reports": ReportsSettings,
             "aws_feature": AWSFeatureSettings,
             "feat_incident": IncidentFeatureSettings,
+            "sre_ops": SreOpsSettings,
             "dev": DevSettings,
         }
 

app/integrations/slack/client.py

@@ -0,0 +1,15 @@
+from slack_sdk import WebClient
+from core.config import settings
+
+
+class SlackClientManager:
+    """Manages the Slack API client. Ensures a single instance is used throughout the application."""
+
+    _client = None
+
+    @classmethod
+    def get_client(cls) -> WebClient:
+        """Returns a singleton instance of the Slack WebClient."""
+        if cls._client is None:
+            cls._client = WebClient(token=settings.slack.SLACK_TOKEN)
+        return cls._client

app/jobs/revoke_aws_sso_access.py

@@ -1,7 +1,7 @@
-from modules.aws import aws_access_requests
-from integrations.aws import sso_admin, identity_store
-from server.utils import log_ops_message
 from core.logging import get_module_logger
+from integrations.aws import identity_store, sso_admin
+from modules.aws import aws_access_requests
+from modules.ops.notifications import log_ops_message
 
 logger = get_module_logger()
 
@@ -37,7 +37,7 @@ def revoke_aws_sso_access(client):
                 user=user_id,
                 text=msg,
             )
-            log_ops_message(client, msg)
+            log_ops_message(msg)
 
         except Exception as e:
             logger.error(

app/modules/aws/aws_access_requests.py

@@ -1,12 +1,11 @@
-import boto3  # type: ignore
 import datetime
 import uuid
 
-from slack_sdk import WebClient
-from server.utils import log_ops_message
+import boto3  # type: ignore
 from core.config import settings
 from integrations.aws import identity_store, organizations, sso_admin
-
+from modules.ops.notifications import log_ops_message
+from slack_sdk import WebClient
 
 PREFIX = settings.PREFIX
 
@@ -200,7 +199,7 @@ def access_view_handler(ack, body, logger, client: WebClient):
         access_type=access_type,
         rationale=rationale,
     )
-    log_ops_message(client, msg)
+    log_ops_message(msg)
     aws_user_id = identity_store.get_user_id(email)
 
     if aws_user_id is None:

app/modules/ops/notifications.py

@@ -0,0 +1,34 @@
+from slack_sdk.errors import SlackApiError
+from core.config import settings
+from core.logging import get_module_logger
+from integrations.slack.client import SlackClientManager
+
+OPS_CHANNEL_ID = settings.sre_ops.SRE_OPS_CHANNEL_ID
+
+logger = get_module_logger()
+
+
+def log_ops_message(message: str):
+    """Provides a standardized way to log operational messages to a specific Slack channel configured in the settings.
+    Failure to log the message will not raise an exception, but will be logged in the application logs to avoid disrupting the main application flow.
+
+    Args:
+        message (str): The message to be logged to the operations channel.
+
+    Returns:
+        None
+    """
+    client = SlackClientManager.get_client()
+    if not client:
+        logger.error("slack_client_not_initialized")
+        return
+    if not OPS_CHANNEL_ID:
+        logger.warning("ops_channel_id_not_configured")
+        return
+    channel_id = OPS_CHANNEL_ID
+    logger.info("ops_message_logged", message=message)
+    try:
+        client.conversations_join(channel=channel_id)
+        client.chat_postMessage(channel=channel_id, text=message, as_user=True)
+    except SlackApiError as e:
+        logger.error("ops_message_failed", message=message, error=str(e))