servo_webhooks.py

from __future__ import annotations
import hmac
import hashlib
import datetime
from importlib.metadata import PackageNotFoundError, version
from typing import Any, Dict, List, Optional

import httpx
import pydantic
import devtools
from pydantic import AnyHttpUrl, BaseModel, Field, SecretStr, validator

import servo
from servo import metadata, License, Maturity, Duration
from servo.events import (
    EventContext,
    EventResult,
    Preposition,
    validate_event_contexts
)


try:
    __version__ = version("servo-webhooks")
except PackageNotFoundError:
    __version__ = "0.0.0"


SUCCESS_STATUS_CODES = (
    httpx.codes.OK,
    httpx.codes.CREATED,
    httpx.codes.ACCEPTED,
    httpx.codes.NO_CONTENT,
    httpx.codes.ALREADY_REPORTED
)
CONTENT_TYPE = "application/vnd.opsani.servo-webhooks+json"


class BackoffConfig(BaseModel):
    """
    The BackoffConfig class provides configuration for backoff and retry provided
    by the backoff library.
    """
    max_time: Duration = '3m'
    max_tries: int = 12


class Webhook(servo.BaseConfiguration):
    name: Optional[str] = Field(
        description="A unique name identifying the webhook.",
    )
    description: Optional[str] = Field(
        description="Optional free-form text describing the context or purpose of the webhook.",
    )
    events: List[str] = Field(
        [],
        description="A list of events that the webhook is listening for.",
    )
    channels: List[servo.pubsub.ChannelName] = Field(
        [],
        description="A list of channels that the webhook is subscribed to.",
    )
    response_channel: Optional[str] = Field(
        description="The channel to publish webhook responses to.",
    )
    url: AnyHttpUrl = Field(
        description="An HTTP, HTTPS, or HTTP/2 endpoint listening for webhooks event requests.",
    )
    secret: SecretStr = Field(
        description="A secret string value used to produce an HMAC digest for verifying webhook authenticity.",
    )
    headers: Dict[str, str] = Field(
        {},
        description="A dictionary of supplemental HTTP headers to include in webhook requests.",
    )
    backoff: BackoffConfig = BackoffConfig()

    @validator("events", "channels", pre=True)
    def _coerce_single_event_to_list(cls, v):
        if isinstance(v, str):
            return [v]
        return v

    # Map strings from config into EventContext objects
    _validate_events = validator("events", pre=True, allow_reuse=True)(validate_event_contexts)

    @pydantic.root_validator(skip_on_failure=True)
    @classmethod
    def _validate_has_events_or_channels(cls, values: dict) -> Dict[str, Any]:
        events, channels = values["events"], values["channels"]
        if len(events) == 0 and len(channels) == 0:
            raise ValueError(
                f"missing webhook data source: events and channels cannot both be empty"
            )

        return values


class WebhooksConfiguration(servo.AbstractBaseConfiguration):
    __root__: List[Webhook] = []

    @classmethod
    def generate(cls, **kwargs) -> "WebhooksConfiguration":
        return cls(
            __root__=[
                Webhook(
                    name="My Webhook",
                    description="Listens for after:measure events and sends an email",
                    events=["after:measure"],
                    url="https://example.com/",
                    secret="s3cr3t!"
                )
            ],
            **kwargs,
        )

    @property
    def webhooks(self) -> List[Webhook]:
        """
        Convenience method for retrieving the root type as a list of webhooks.
        """
        return self.__root__


class Result(BaseModel):
    """Models an EventResult webhook representation"""
    connector: str
    value: Any


class RequestBody(BaseModel):
    """Models the JSON body of a webhook request containing event results"""
    event: str
    created_at: datetime.datetime = pydantic.Field(default_factory=datetime.datetime.now)
    results: Optional[List[Result]] = None


@metadata(
    name=("servo-webhooks", "webhooks"),
    description="Dispatch servo events via HTTP webhooks",
    version=__version__,
    homepage="https://github.com/opsani/servo-webhooks",
    license=License.apache2,
    maturity=Maturity.experimental,
)
class WebhooksConnector(servo.BaseConnector):
    config: WebhooksConfiguration

    @servo.on_event()
    async def startup(self) -> None:
        self._register_event_handlers()
        self._register_subscriber_handlers()

    ##
    # Event Webhooks

    def _register_event_handlers(self) -> None:
        for webhook in self.config.webhooks:
            for event_name in webhook.events:
                event = EventContext.from_str(event_name)
                if not event:
                    raise ValueError(f"invalid webhook event '{event_name}'")
                if event.preposition == Preposition.before:
                    self._add_before_event_webhook_handler(webhook, event)
                elif event.preposition == Preposition.after:
                    self._add_after_event_webhook_handler(webhook, event)
                else:
                    raise ValueError(f"Unsupported Preposition value given for webhook: '{event.preposition}'")

    def _add_before_event_webhook_handler(self, webhook: Webhook, event: EventContext) -> None:
        async def __before_handler(self, **kwargs) -> None:
            headers = {**webhook.headers, **{ "Content-Type": CONTENT_TYPE }}
            async with httpx.AsyncClient(headers=headers) as client:
                body = RequestBody(event=str(event))
                json_body = body.json()
                headers["X-Servo-Signature"] = _signature_for_webhook_body(webhook, json_body)
                async with httpx.AsyncClient(headers=headers) as client:
                    try:
                        response = await client.post(webhook.url, data=json_body, headers=headers)
                        success = (response.status_code in SUCCESS_STATUS_CODES)
                        if success:
                            self.logger.success(f"posted webhook for '{event}' event to '{webhook.url}' ({response.status_code} {response.reason_phrase})")
                        else:
                            self.logger.error(f"failed posted webhook for '{event}' event to '{webhook.url}' ({response.status_code} {response.reason_phrase}): {response.text}")

                        await self._publish_response_if_necessary(response, webhook)

                    except (httpx.RequestError, httpx.HTTPError) as error:
                        self.logger.error(f"HTTP error \"{error.__class__.__name__}\" encountered while posting to webhook \"{webhook.url}\": {error}")
                        self.logger.trace(f"Webhook: {devtools.pformat(webhook)}, request body={devtools.pformat(json_body)}")

        self.add_event_handler(event.event, event.preposition, __before_handler)

    def _add_after_event_webhook_handler(self, webhook: Webhook, event: EventContext) -> None:
        async def __after_handler(self, results: List[EventResult], **kwargs) -> None:
            headers = {**webhook.headers, **{ "Content-Type": CONTENT_TYPE }}

            outbound_results = []
            for result in results:
                outbound_results.append(
                    Result(
                        connector=result.connector.name,
                        value=result.value
                    )
                )
            body = RequestBody(
                event=str(event),
                results=outbound_results
            )

            json_body = body.json()
            headers["X-Servo-Signature"] = _signature_for_webhook_body(webhook, json_body)
            async with httpx.AsyncClient(headers=headers) as client:
                try:
                    response = await client.post(webhook.url, data=json_body, headers=headers)
                    success = (response.status_code in SUCCESS_STATUS_CODES)
                    if success:
                        self.logger.success(f"posted webhook for '{event}' event to '{webhook.url}' ({response.status_code} {response.reason_phrase})")
                    else:
                        self.logger.error(f"failed posted webhook for '{event}' event to '{webhook.url}' ({response.status_code} {response.reason_phrase}): {response.text}")

                    await self._publish_response_if_necessary(response, webhook)

                except (httpx.RequestError, httpx.HTTPError) as error:
                    self.logger.error(f"HTTP error \"{error.__class__.__name__}\" encountered while posting to webhook \"{webhook.url}\": {error}")
                    self.logger.trace(f"Webhook: {devtools.pformat(webhook)}, request body={devtools.pformat(json_body)}")

        self.add_event_handler(event.event, event.preposition, __after_handler)

    ##
    # Pub/sub Webhooks

    def _register_subscriber_handlers(self) -> None:
        for webhook in self.config.webhooks:
            for channel in webhook.channels:
                self._add_subscriber_for_webhook_channel(webhook, channel)

    def _add_subscriber_for_webhook_channel(self, webhook: Webhook, channel: str) -> None:
        @self.subscribe(channel)
        async def _message_received(message: servo.Message, channel: servo.Channel) -> None:
            servo.logger.debug(f"Notified of a new Message: {message}, {channel}")

            headers = {**webhook.headers, **{ "Content-Type": CONTENT_TYPE }}
            headers["X-Servo-Signature"] = _signature_for_webhook_body(webhook, message.text)
            async with httpx.AsyncClient(headers=headers) as client:
                try:
                    response = await client.post(webhook.url, data=message.content, headers=headers)
                    success = (response.status_code in SUCCESS_STATUS_CODES)
                    if success:
                        self.logger.success(f"posted webhook for message sent to '{channel}' sent to '{webhook.url}' ({response.status_code} {response.reason_phrase})")
                    else:
                        self.logger.error(f"failed posted webhook for message sent to '{channel}' event to '{webhook.url}' ({response.status_code} {response.reason_phrase}): {response.text}")

                    await self._publish_response_if_necessary(response, webhook)

                except (httpx.RequestError, httpx.HTTPError) as error:
                    self.logger.error(f"HTTP error \"{error.__class__.__name__}\" encountered while posting to webhook \"{webhook.url}\": {error}")
                    self.logger.trace(f"Webhook: {devtools.pformat(webhook)}, request body={devtools.pformat(json_body)}")

    async def _publish_response_if_necessary(self, response: httpx.Response, webhook: Webhook) -> None:
        # Only publish successful responses for the moment
        if not response.status_code in SUCCESS_STATUS_CODES:
            return

        content_type = response.headers.get("Content-Type")
        if content_type is None:
            return

        if len(response.content) and webhook.response_channel:
            async with self.publish(webhook.response_channel) as publisher:
                message = servo.Message(content=response.content, content_type=content_type)
                self.logger.info(f"Publishing response message to channel '{webhook.response_channel}': {message}")
                await publisher(message)


def _signature_for_webhook_body(webhook: Webhook, body: str) -> str:
    secret_bytes = webhook.secret.get_secret_value().encode()
    return str(hmac.new(secret_bytes, body.encode(), hashlib.sha1).hexdigest())


# class CLI(servo.cli.ConnectorCLI):
#     pass
#     # list, add, remove, test