authentik.events.api.notification_transports

  1"""NotificationTransport API Views"""
  2
  3from typing import Any
  4
  5from drf_spectacular.types import OpenApiTypes
  6from drf_spectacular.utils import OpenApiResponse, extend_schema
  7from rest_framework.decorators import action
  8from rest_framework.exceptions import ValidationError
  9from rest_framework.fields import CharField, ListField, SerializerMethodField
 10from rest_framework.request import Request
 11from rest_framework.response import Response
 12from rest_framework.viewsets import ModelViewSet
 13
 14from authentik.core.api.used_by import UsedByMixin
 15from authentik.core.api.utils import ModelSerializer, PassiveSerializer
 16from authentik.events.models import (
 17    Event,
 18    Notification,
 19    NotificationSeverity,
 20    NotificationTransport,
 21    NotificationTransportError,
 22    TransportMode,
 23)
 24from authentik.events.utils import get_user
 25from authentik.rbac.decorators import permission_required
 26from authentik.stages.email.models import get_template_choices
 27
 28
 29class NotificationTransportSerializer(ModelSerializer):
 30    """NotificationTransport Serializer"""
 31
 32    mode_verbose = SerializerMethodField()
 33
 34    def __init__(self, *args, **kwargs):
 35        super().__init__(*args, **kwargs)
 36        self.fields["email_template"].choices = get_template_choices()
 37
 38    def validate_email_template(self, value: str) -> str:
 39        """Check validity of email template"""
 40        choices = get_template_choices()
 41        for path, _ in choices:
 42            if path == value:
 43                return value
 44        raise ValidationError(f"Invalid template '{value}' specified.")
 45
 46    def get_mode_verbose(self, instance: NotificationTransport) -> str:
 47        """Return selected mode with a UI Label"""
 48        return TransportMode(instance.mode).label
 49
 50    def validate(self, attrs: dict[Any, str]) -> dict[Any, str]:
 51        """Ensure the required fields are set."""
 52        mode = attrs.get("mode")
 53        if mode in [TransportMode.WEBHOOK, TransportMode.WEBHOOK_SLACK]:
 54            if "webhook_url" not in attrs or attrs.get("webhook_url", "") == "":
 55                raise ValidationError({"webhook_url": "Webhook URL may not be empty."})
 56        return attrs
 57
 58    class Meta:
 59        model = NotificationTransport
 60        fields = [
 61            "pk",
 62            "name",
 63            "mode",
 64            "mode_verbose",
 65            "webhook_url",
 66            "webhook_ca",
 67            "webhook_mapping_body",
 68            "webhook_mapping_headers",
 69            "email_subject_prefix",
 70            "email_template",
 71            "send_once",
 72        ]
 73
 74
 75class NotificationTransportTestSerializer(PassiveSerializer):
 76    """Notification test serializer"""
 77
 78    messages = ListField(child=CharField())
 79
 80
 81class NotificationTransportViewSet(UsedByMixin, ModelViewSet):
 82    """NotificationTransport Viewset"""
 83
 84    queryset = NotificationTransport.objects.all()
 85    serializer_class = NotificationTransportSerializer
 86    filterset_fields = ["name", "mode", "webhook_url", "send_once"]
 87    search_fields = ["name", "mode", "webhook_url"]
 88    ordering = ["name"]
 89
 90    @permission_required("authentik_events.change_notificationtransport")
 91    @extend_schema(
 92        responses={
 93            200: NotificationTransportTestSerializer(many=False),
 94            500: OpenApiResponse(description="Failed to test transport"),
 95        },
 96        request=OpenApiTypes.NONE,
 97    )
 98    @action(detail=True, pagination_class=None, filter_backends=[], methods=["post"])
 99    def test(self, request: Request, pk=None) -> Response:
100        """Send example notification using selected transport. Requires
101        Modify permissions."""
102        transport: NotificationTransport = self.get_object()
103        event = Event.new(
104            action="notification_test",
105            user=get_user(request.user),
106            context={"foo": "bar"},
107        )
108        event.save()
109        notification = Notification(
110            severity=NotificationSeverity.NOTICE,
111            body=f"Test Notification from transport {transport.name}",
112            user=request.user,
113            event=event,
114        )
115        try:
116            response = NotificationTransportTestSerializer(
117                data={"messages": transport.send(notification)}
118            )
119            response.is_valid()
120            return Response(response.data)
121        except NotificationTransportError as exc:
122            return Response(str(exc.__cause__ or None), status=500)