authentik.enterprise.api

  1"""Enterprise API Views"""
  2
  3from collections.abc import Callable
  4from datetime import timedelta
  5from functools import wraps
  6
  7from django.utils.timezone import now
  8from django.utils.translation import gettext as _
  9from drf_spectacular.types import OpenApiTypes
 10from drf_spectacular.utils import OpenApiParameter, extend_schema, inline_serializer
 11from rest_framework.decorators import action
 12from rest_framework.exceptions import ValidationError
 13from rest_framework.fields import CharField, IntegerField
 14from rest_framework.permissions import IsAuthenticated
 15from rest_framework.request import Request
 16from rest_framework.response import Response
 17from rest_framework.validators import UniqueValidator
 18from rest_framework.viewsets import ModelViewSet
 19
 20from authentik.core.api.used_by import UsedByMixin
 21from authentik.core.api.utils import ModelSerializer, PassiveSerializer
 22from authentik.core.models import User, UserTypes
 23from authentik.enterprise.license import LicenseKey, LicenseSummarySerializer
 24from authentik.enterprise.models import License
 25from authentik.rbac.decorators import permission_required
 26from authentik.tenants.utils import get_unique_identifier
 27
 28
 29class EnterpriseRequiredMixin:
 30    """Mixin to validate that a valid enterprise license
 31    exists before allowing to save the object"""
 32
 33    def validate(self, attrs: dict) -> dict:
 34        """Check that a valid license exists"""
 35        if not LicenseKey.cached_summary().status.is_valid:
 36            raise ValidationError(_("Enterprise is required to create/update this object."))
 37        return super().validate(attrs)
 38
 39
 40def enterprise_action(func: Callable):
 41    """Check permissions for a single custom action"""
 42
 43    @wraps(func)
 44    def wrapper(*args, **kwargs) -> Response:
 45        if not LicenseKey.cached_summary().status.is_valid:
 46            raise ValidationError(_("Enterprise is required to use this endpoint."))
 47        return func(*args, **kwargs)
 48
 49    return wrapper
 50
 51
 52class LicenseSerializer(ModelSerializer):
 53    """License Serializer"""
 54
 55    def validate_key(self, key: str) -> str:
 56        """Validate the license key (install_id and signature)"""
 57        LicenseKey.validate(key)
 58        return key
 59
 60    class Meta:
 61        model = License
 62        fields = [
 63            "license_uuid",
 64            "name",
 65            "key",
 66            "expiry",
 67            "internal_users",
 68            "external_users",
 69        ]
 70        extra_kwargs = {
 71            "key": {"validators": [UniqueValidator(queryset=License.objects.all())]},
 72            "name": {"read_only": True},
 73            "expiry": {"read_only": True},
 74            "internal_users": {"read_only": True},
 75            "external_users": {"read_only": True},
 76        }
 77
 78
 79class LicenseForecastSerializer(PassiveSerializer):
 80    """Serializer for license forecast"""
 81
 82    internal_users = IntegerField(required=True)
 83    external_users = IntegerField(required=True)
 84    forecasted_internal_users = IntegerField(required=True)
 85    forecasted_external_users = IntegerField(required=True)
 86
 87
 88class LicenseViewSet(UsedByMixin, ModelViewSet):
 89    """License Viewset"""
 90
 91    queryset = License.objects.all()
 92    serializer_class = LicenseSerializer
 93    search_fields = ["name"]
 94    ordering = ["name"]
 95    filterset_fields = ["name"]
 96
 97    @permission_required(None, ["authentik_enterprise.view_license"])
 98    @extend_schema(
 99        request=OpenApiTypes.NONE,
100        responses={
101            200: inline_serializer("InstallIDSerializer", {"install_id": CharField(required=True)}),
102        },
103    )
104    @action(detail=False, methods=["GET"])
105    def install_id(self, request: Request) -> Response:
106        """Get install_id"""
107        return Response(
108            data={
109                "install_id": get_unique_identifier(),
110            }
111        )
112
113    @extend_schema(
114        request=OpenApiTypes.NONE,
115        responses={
116            200: LicenseSummarySerializer(),
117        },
118        parameters=[
119            OpenApiParameter(
120                name="cached",
121                location=OpenApiParameter.QUERY,
122                type=OpenApiTypes.BOOL,
123                default=True,
124            )
125        ],
126    )
127    @action(detail=False, methods=["GET"], permission_classes=[IsAuthenticated])
128    def summary(self, request: Request) -> Response:
129        """Get the total license status"""
130        summary = LicenseKey.cached_summary()
131        if request.query_params.get("cached", "true").lower() == "false":
132            summary = LicenseKey.get_total().summary()
133        response = LicenseSummarySerializer(instance=summary)
134        return Response(response.data)
135
136    @permission_required(None, ["authentik_enterprise.view_license"])
137    @extend_schema(
138        request=OpenApiTypes.NONE,
139        responses={
140            200: LicenseForecastSerializer(),
141        },
142    )
143    @action(detail=False, methods=["GET"])
144    def forecast(self, request: Request) -> Response:
145        """Forecast how many users will be required in a year"""
146        last_month = now() - timedelta(days=30)
147        # Forecast for internal users
148        internal_in_last_month = User.objects.filter(
149            type=UserTypes.INTERNAL, date_joined__gte=last_month
150        ).count()
151        # Forecast for external users
152        external_in_last_month = LicenseKey.get_external_user_count()
153        forecast_for_months = 12
154        response = LicenseForecastSerializer(
155            data={
156                "internal_users": LicenseKey.get_internal_user_count(),
157                "external_users": LicenseKey.get_external_user_count(),
158                "forecasted_internal_users": (internal_in_last_month * forecast_for_months),
159                "forecasted_external_users": (external_in_last_month * forecast_for_months),
160            }
161        )
162        response.is_valid(raise_exception=True)
163        return Response(response.data)