ACLP : Webhook alert channels

linode_api4/groups/monitor.py

@@ -20,8 +20,14 @@
 )
 from linode_api4.objects.monitor import (
     AkamaiObjectStorageLogsDestinationDetails,
+    BasicAuthenticationDetails,
+    ChannelDetails,
+    CustomHeader,
     CustomHTTPSLogsDestinationDetails,
+    DestinationAuthentication,
+    EmailDetails,
     LogsStreamDetails,
+    WebhookDetails,
 )
 
 __all__ = [
@@ -213,6 +219,193 @@ def alert_channels(self, *filters) -> PaginatedList:
         """
         return self.client._get_and_filter(AlertChannel, *filters)
 
+    def channel_create(
+        self,
+        label: str,
+        channel_type: str,
+        details: "ChannelDetails",
+    ) -> AlertChannel:
+        """
+        Create a new alert channel.
+
+        Alert channels define destinations for alert notifications. Supported
+        channel types include email, webhook, PagerDuty, and Slack.
+
+        **Webhook Channel Constraints:**
+        - If ``channel_type`` is "webhook", the following are required:
+          - ``details.webhook.endpoint_url`` must be provided
+          - ``details.webhook.authentication.type`` must be specified ("basic" or "none")
+          - If ``authentication.type`` is "basic", both ``basic_authentication_user`` and
+            ``basic_authentication_password`` must be provided in ``details.webhook.authentication.details``
+        - Client Certificate Configuration (Optional but must be complete):
+          - If ``details.webhook.client_certificate_details`` is provided, all three certificates
+            must be included: ``client_ca_certificate``, ``client_certificate``, and ``client_private_key``
+          - ``tls_hostname`` is optional
+        - Custom Headers:
+          - ``Content-Type`` header must NOT be set by the user; it will be managed by the API
+
+        API Documentation: https://techdocs.akamai.com/linode-api/reference/post-notification-channel
+
+        :param label: A human-readable name for the alert channel.
+        :type label: str
+        :param channel_type: The channel type (e.g., ``"email"``, ``"webhook"``).
+        :type channel_type: str
+        :param details: Configuration details specific to the channel type.
+        :type details: ChannelDetails
+
+        :returns: The newly created alert channel.
+        :rtype: AlertChannel
+
+        :raises ValueError: If webhook channel configuration is invalid or missing required fields.
+        """
+        pass
+
+        # Validate webhook channel requirements
+        if channel_type == "webhook":
+            self._validate_webhook_details(details)
+
+        params = {
+            "label": label,
+            "channel_type": channel_type,
+            "details": (
+                details._serialize()
+                if hasattr(details, "_serialize")
+                else details
+            ),
+        }
+
+        result = self.client.post("/monitor/alert-channels", data=params)
+
+        if "id" not in result:
+            raise UnexpectedResponseError(
+                "Unexpected response when creating alert channel!",
+                json=result,
+            )
+
+        return AlertChannel(self.client, result["id"], result)
+
+    def verify_webhook(
+        self,
+        webhook: "WebhookDetails",
+    ) -> bool:
+        """
+        Verify a webhook configuration by testing connectivity to the endpoint.
+
+        This endpoint validates that the webhook endpoint is reachable and
+        accepts the request format. It's recommended to verify webhook
+        configurations before creating a webhook channel.
+
+        **Webhook Configuration Requirements:**
+        - ``endpoint_url`` must be provided
+        - ``authentication.type`` must be specified ("basic" or "none")
+        - If ``authentication.type`` is "basic", both ``basic_authentication_user`` and
+          ``basic_authentication_password`` must be provided
+        - Client certificates (if used) must include all three: ``client_ca_certificate``,
+          ``client_certificate``, and ``client_private_key``
+
+        API Documentation: https://techdocs.akamai.com/linode-api/reference/post-verify-webhook
+
+        :param webhook: The webhook configuration to verify.
+        :type webhook: WebhookDetails
+
+        :returns: True if verification succeeds.
+        :rtype: bool
+
+        :raises ValueError: If webhook configuration is invalid.
+        :raises ApiError: If the webhook verification fails.
+        """
+        from linode_api4.objects.monitor import ChannelDetails
+
+        # Validate webhook configuration
+        self._validate_webhook_details(ChannelDetails(webhook=webhook))
+
+        data = {
+            "webhook": (
+                webhook._serialize()
+                if hasattr(webhook, "_serialize")
+                else webhook
+            ),
+        }
+
+        result = self.client.post("/monitor/alert-channels/verify", data=data)
+
+        return result.get("success", True)
+
+    def _validate_webhook_details(self, details: "ChannelDetails") -> None:
+        """
+        Validate webhook channel details against API requirements.
+
+        :param details: The channel details to validate.
+        :type details: ChannelDetails
+
+        :raises ValueError: If validation fails.
+        """
+        if not details or not details.webhook:
+            raise ValueError(
+                "Webhook details are required for webhook channel type"
+            )
+
+        webhook = details.webhook
+
+        # Validate required endpoint_url
+        if not webhook.endpoint_url:
+            raise ValueError(
+                "Webhook channel requires 'endpoint_url' to be specified"
+            )
+
+        # Validate required authentication.type
+        if not webhook.authentication or not webhook.authentication.type:
+            raise ValueError(
+                "Webhook channel requires 'authentication.type' to be specified "
+                "(e.g., 'basic' or 'none')"
+            )
+
+        auth_type = webhook.authentication.type
+        if auth_type == "basic":
+            # For basic auth, both username and password are required
+            if not webhook.authentication.details:
+                raise ValueError(
+                    "Basic authentication requires 'authentication.details' to be specified"
+                )
+
+            auth_details = webhook.authentication.details
+            if not auth_details.basic_authentication_user:
+                raise ValueError(
+                    "Basic authentication requires 'basic_authentication_user' to be specified"
+                )
+
+            if not auth_details.basic_authentication_password:
+                raise ValueError(
+                    "Basic authentication requires 'basic_authentication_password' to be specified"
+                )
+
+        # Validate client certificate configuration (all three must be present together)
+        if webhook.client_certificate_details:
+            cert_details = webhook.client_certificate_details
+
+            # Check if any certificate field is present
+            has_ca_cert = bool(cert_details.client_ca_certificate)
+            has_client_cert = bool(cert_details.client_certificate)
+            has_private_key = bool(cert_details.client_private_key)
+
+            # If any certificate field is present, all must be present
+            if has_ca_cert or has_client_cert or has_private_key:
+                if not (has_ca_cert and has_client_cert and has_private_key):
+                    raise ValueError(
+                        "Client certificate configuration requires all three to be specified: "
+                        "'client_ca_certificate', 'client_certificate', and 'client_private_key'. "
+                        "'tls_hostname' is optional."
+                    )
+
+        # Validate custom headers don't include Content-Type
+        if webhook.custom_headers:
+            for header in webhook.custom_headers:
+                if header.name and header.name.lower() == "content-type":
+                    raise ValueError(
+                        "Custom headers must NOT include 'Content-Type'; "
+                        "it will be managed by the API"
+                    )
+
     def create_alert_definition(
         self,
         service_type: str,