authentik.crypto.api

  1"""Crypto API Views"""
  2
  3from cryptography.hazmat.backends import default_backend
  4from cryptography.hazmat.primitives.serialization import load_pem_private_key
  5from cryptography.x509 import load_pem_x509_certificate
  6from django.http.response import HttpResponse
  7from django.urls import reverse
  8from django.utils.translation import gettext_lazy as _
  9from django_filters import FilterSet
 10from django_filters.filters import BooleanFilter, MultipleChoiceFilter
 11from drf_spectacular.types import OpenApiTypes
 12from drf_spectacular.utils import (
 13    OpenApiParameter,
 14    OpenApiResponse,
 15    extend_schema,
 16)
 17from rest_framework.decorators import action
 18from rest_framework.exceptions import ValidationError
 19from rest_framework.fields import (
 20    CharField,
 21    ChoiceField,
 22    IntegerField,
 23    SerializerMethodField,
 24)
 25from rest_framework.filters import OrderingFilter, SearchFilter
 26from rest_framework.permissions import IsAuthenticated
 27from rest_framework.request import Request
 28from rest_framework.response import Response
 29from rest_framework.validators import UniqueValidator
 30from rest_framework.viewsets import ModelViewSet
 31from structlog.stdlib import get_logger
 32
 33from authentik.api.validation import validate
 34from authentik.core.api.used_by import UsedByMixin
 35from authentik.core.api.utils import ModelSerializer, PassiveSerializer
 36from authentik.core.models import UserTypes
 37from authentik.crypto.apps import MANAGED_KEY
 38from authentik.crypto.builder import CertificateBuilder, PrivateKeyAlg
 39from authentik.crypto.models import CertificateKeyPair, KeyType
 40from authentik.events.models import Event, EventAction
 41from authentik.rbac.decorators import permission_required
 42from authentik.rbac.filters import SecretKeyFilter
 43
 44LOGGER = get_logger()
 45
 46
 47class CertificateKeyPairSerializer(ModelSerializer):
 48    """CertificateKeyPair Serializer"""
 49
 50    private_key_available = SerializerMethodField()
 51
 52    certificate_download_url = SerializerMethodField()
 53    private_key_download_url = SerializerMethodField()
 54
 55    def get_private_key_available(self, instance: CertificateKeyPair) -> bool:
 56        """Show if this keypair has a private key configured or not"""
 57        return instance.key_data != "" and instance.key_data is not None
 58
 59    def get_certificate_download_url(self, instance: CertificateKeyPair) -> str:
 60        """Get URL to download certificate"""
 61        return (
 62            reverse(
 63                "authentik_api:certificatekeypair-view-certificate",
 64                kwargs={"pk": instance.pk},
 65            )
 66            + "?download"
 67        )
 68
 69    def get_private_key_download_url(self, instance: CertificateKeyPair) -> str:
 70        """Get URL to download private key"""
 71        return (
 72            reverse(
 73                "authentik_api:certificatekeypair-view-private-key",
 74                kwargs={"pk": instance.pk},
 75            )
 76            + "?download"
 77        )
 78
 79    def validate_certificate_data(self, value: str) -> str:
 80        """Verify that input is a valid PEM x509 Certificate"""
 81        try:
 82            # Cast to string to fully load and parse certificate
 83            # Prevents issues like https://github.com/goauthentik/authentik/issues/2082
 84            str(load_pem_x509_certificate(value.encode("utf-8"), default_backend()))
 85        except ValueError as exc:
 86            LOGGER.warning("Failed to load certificate", exc=exc)
 87            raise ValidationError("Unable to load certificate.") from None
 88        return value
 89
 90    def validate_key_data(self, value: str) -> str:
 91        """Verify that input is a valid PEM Key"""
 92        # Since this field is optional, data can be empty.
 93        if value != "":
 94            try:
 95                # Cast to string to fully load and parse certificate
 96                # Prevents issues like https://github.com/goauthentik/authentik/issues/2082
 97                str(
 98                    load_pem_private_key(
 99                        str.encode("\n".join([x.strip() for x in value.split("\n")])),
100                        password=None,
101                        backend=default_backend(),
102                    )
103                )
104            except (ValueError, TypeError) as exc:
105                LOGGER.warning("Failed to load private key", exc=exc)
106                raise ValidationError("Unable to load private key (possibly encrypted?).") from None
107        return value
108
109    class Meta:
110        model = CertificateKeyPair
111        fields = [
112            "pk",
113            "name",
114            "fingerprint_sha256",
115            "fingerprint_sha1",
116            "certificate_data",
117            "key_data",
118            "cert_expiry",
119            "cert_subject",
120            "private_key_available",
121            "key_type",
122            "certificate_download_url",
123            "private_key_download_url",
124            "managed",
125        ]
126        extra_kwargs = {
127            "managed": {"read_only": True},
128            "key_data": {"write_only": True},
129            "certificate_data": {"write_only": True},
130            "fingerprint_sha256": {"read_only": True},
131            "fingerprint_sha1": {"read_only": True},
132            "cert_expiry": {"read_only": True},
133            "cert_subject": {"read_only": True},
134            "key_type": {"read_only": True},
135        }
136
137
138class CertificateDataSerializer(PassiveSerializer):
139    """Get CertificateKeyPair's data"""
140
141    data = CharField(read_only=True)
142
143
144class CertificateGenerationSerializer(PassiveSerializer):
145    """Certificate generation parameters"""
146
147    common_name = CharField(
148        validators=[UniqueValidator(queryset=CertificateKeyPair.objects.all())],
149        source="name",
150    )
151    subject_alt_name = CharField(required=False, allow_blank=True, label=_("Subject-alt name"))
152    validity_days = IntegerField(initial=365)
153    alg = ChoiceField(default=PrivateKeyAlg.RSA, choices=PrivateKeyAlg.choices)
154
155
156class CertificateKeyPairFilter(FilterSet):
157    """Filter for certificates"""
158
159    has_key = BooleanFilter(
160        label="Only return certificate-key pairs with keys", method="filter_has_key"
161    )
162
163    key_type = MultipleChoiceFilter(
164        choices=KeyType.choices,
165        label="Filter by key algorithm type",
166        method="filter_key_type",
167    )
168
169    def filter_has_key(self, queryset, name, value):  # pragma: no cover
170        """Only return certificate-key pairs with keys"""
171        if not value:
172            return queryset
173        return queryset.exclude(key_data__exact="")
174
175    def filter_key_type(self, queryset, name, value):  # pragma: no cover
176        """Filter certificates by key type using the stored database field"""
177        if not value:
178            return queryset
179
180        # value is a list of KeyType enum values from MultipleChoiceFilter
181        return queryset.filter(key_type__in=value)
182
183    class Meta:
184        model = CertificateKeyPair
185        fields = ["name", "managed"]
186
187
188class CertificateKeyPairViewSet(UsedByMixin, ModelViewSet):
189    """CertificateKeyPair Viewset"""
190
191    queryset = CertificateKeyPair.objects.exclude(managed=MANAGED_KEY)
192    serializer_class = CertificateKeyPairSerializer
193    filterset_class = CertificateKeyPairFilter
194    ordering = ["name"]
195    search_fields = ["name"]
196    filter_backends = [SecretKeyFilter, OrderingFilter, SearchFilter]
197
198    @extend_schema(
199        parameters=[
200            # Override the type for `has_key` above
201            OpenApiParameter(
202                "has_key",
203                bool,
204                required=False,
205                description="Only return certificate-key pairs with keys",
206            ),
207            OpenApiParameter(
208                "key_type",
209                OpenApiTypes.STR,
210                required=False,
211                many=True,
212                enum=[choice[0] for choice in KeyType.choices],
213                description=(
214                    "Filter by key algorithm type (RSA, EC, DSA, etc). "
215                    "Can be specified multiple times (e.g. '?key_type=rsa&key_type=ec')"
216                ),
217            ),
218        ]
219    )
220    def list(self, request, *args, **kwargs):
221        return super().list(request, *args, **kwargs)
222
223    @permission_required(None, ["authentik_crypto.add_certificatekeypair"])
224    @extend_schema(
225        request=CertificateGenerationSerializer(),
226        responses={
227            200: CertificateKeyPairSerializer,
228            400: OpenApiResponse(description="Bad request"),
229        },
230    )
231    @action(detail=False, methods=["POST"])
232    @validate(CertificateGenerationSerializer)
233    def generate(self, request: Request, body: CertificateGenerationSerializer) -> Response:
234        """Generate a new, self-signed certificate-key pair"""
235        raw_san = body.validated_data.get("subject_alt_name", "")
236        sans = raw_san.split(",") if raw_san != "" else []
237        builder = CertificateBuilder(body.validated_data["name"])
238        builder.alg = body.validated_data["alg"]
239        builder.build(
240            subject_alt_names=sans,
241            validity_days=int(body.validated_data["validity_days"]),
242        )
243        instance = builder.save()
244        serializer = self.get_serializer(instance)
245        return Response(serializer.data)
246
247    @permission_required("view_certificatekeypair_certificate")
248    @extend_schema(
249        parameters=[
250            OpenApiParameter(
251                name="download",
252                location=OpenApiParameter.QUERY,
253                type=OpenApiTypes.BOOL,
254            )
255        ],
256        responses={200: CertificateDataSerializer(many=False)},
257    )
258    @action(detail=True, pagination_class=None, permission_classes=[IsAuthenticated])
259    def view_certificate(self, request: Request, pk: str) -> Response:
260        """Return certificate-key pairs certificate and log access"""
261        certificate: CertificateKeyPair = self.get_object()
262        if request.user.type != UserTypes.INTERNAL_SERVICE_ACCOUNT:
263            Event.new(  # noqa # nosec
264                EventAction.SECRET_VIEW,
265                secret=certificate,
266                type="certificate",
267            ).from_http(request)
268        if "download" in request.query_params:
269            # Mime type from https://pki-tutorial.readthedocs.io/en/latest/mime.html
270            response = HttpResponse(
271                certificate.certificate_data, content_type="application/x-pem-file"
272            )
273            response["Content-Disposition"] = (
274                f'attachment; filename="{certificate.name}_certificate.pem"'
275            )
276            return response
277        return Response(CertificateDataSerializer({"data": certificate.certificate_data}).data)
278
279    @permission_required("view_certificatekeypair_key")
280    @extend_schema(
281        parameters=[
282            OpenApiParameter(
283                name="download",
284                location=OpenApiParameter.QUERY,
285                type=OpenApiTypes.BOOL,
286            )
287        ],
288        responses={200: CertificateDataSerializer(many=False)},
289    )
290    @action(detail=True, pagination_class=None, permission_classes=[IsAuthenticated])
291    def view_private_key(self, request: Request, pk: str) -> Response:
292        """Return certificate-key pairs private key and log access"""
293        certificate: CertificateKeyPair = self.get_object()
294        if request.user.type != UserTypes.INTERNAL_SERVICE_ACCOUNT:
295            Event.new(  # noqa # nosec
296                EventAction.SECRET_VIEW,
297                secret=certificate,
298                type="private_key",
299            ).from_http(request)
300        if "download" in request.query_params:
301            # Mime type from https://pki-tutorial.readthedocs.io/en/latest/mime.html
302            response = HttpResponse(certificate.key_data, content_type="application/x-pem-file")
303            response["Content-Disposition"] = (
304                f'attachment; filename="{certificate.name}_private_key.pem"'
305            )
306            return response
307        return Response(CertificateDataSerializer({"data": certificate.key_data}).data)