authentik.providers.proxy.api

ProxyProvider API Views

View Source

  1"""ProxyProvider API Views"""
  2
  3from typing import Any
  4
  5from django.utils.translation import gettext_lazy as _
  6from drf_spectacular.utils import extend_schema_field
  7from rest_framework.exceptions import ValidationError
  8from rest_framework.fields import CharField, ListField, ReadOnlyField, SerializerMethodField
  9from rest_framework.mixins import ListModelMixin
 10from rest_framework.viewsets import GenericViewSet, ModelViewSet
 11
 12from authentik.core.api.providers import ProviderSerializer
 13from authentik.core.api.used_by import UsedByMixin
 14from authentik.core.api.utils import ModelSerializer, PassiveSerializer
 15from authentik.lib.utils.time import timedelta_from_string
 16from authentik.providers.oauth2.api.providers import RedirectURISerializer
 17from authentik.providers.oauth2.models import ScopeMapping
 18from authentik.providers.oauth2.views.provider import ProviderInfoView
 19from authentik.providers.proxy.models import ProxyMode, ProxyProvider
 20
 21
 22class OpenIDConnectConfigurationSerializer(PassiveSerializer):
 23    """rest_framework Serializer for OIDC Configuration"""
 24
 25    issuer = CharField()
 26    authorization_endpoint = CharField()
 27    token_endpoint = CharField()
 28    userinfo_endpoint = CharField()
 29    end_session_endpoint = CharField()
 30    introspection_endpoint = CharField()
 31    jwks_uri = CharField()
 32
 33    response_types_supported = ListField(child=CharField())
 34    id_token_signing_alg_values_supported = ListField(child=CharField())
 35    subject_types_supported = ListField(child=CharField())
 36    token_endpoint_auth_methods_supported = ListField(child=CharField())
 37
 38
 39class ProxyProviderSerializer(ProviderSerializer):
 40    """ProxyProvider Serializer"""
 41
 42    client_id = CharField(read_only=True)
 43    redirect_uris = RedirectURISerializer(many=True, read_only=True, source="_redirect_uris")
 44    outpost_set = ListField(child=CharField(), read_only=True, source="outpost_set.all")
 45
 46    def validate_basic_auth_enabled(self, value: bool) -> bool:
 47        """Ensure user and password attributes are set"""
 48        if value:
 49            if (
 50                self.initial_data.get("basic_auth_password_attribute", "") == ""
 51                or self.initial_data.get("basic_auth_user_attribute", "") == ""
 52            ):
 53                raise ValidationError(
 54                    _("User and password attributes must be set when basic auth is enabled.")
 55                )
 56        return value
 57
 58    def validate(self, attrs) -> dict[Any, str]:
 59        """Check that internal_host is set when mode is Proxy"""
 60        if (
 61            attrs.get("mode", ProxyMode.PROXY) == ProxyMode.PROXY
 62            and attrs.get("internal_host", "") == ""
 63        ):
 64            raise ValidationError(
 65                {"internal_host": _("Internal host cannot be empty when forward auth is disabled.")}
 66            )
 67        return attrs
 68
 69    def create(self, validated_data: dict):
 70        instance: ProxyProvider = super().create(validated_data)
 71        instance.set_oauth_defaults()
 72        instance.save()
 73        return instance
 74
 75    def update(self, instance: ProxyProvider, validated_data: dict):
 76        instance = super().update(instance, validated_data)
 77        instance.set_oauth_defaults()
 78        instance.save()
 79        return instance
 80
 81    class Meta:
 82        model = ProxyProvider
 83        fields = ProviderSerializer.Meta.fields + [
 84            "client_id",
 85            "internal_host",
 86            "external_host",
 87            "internal_host_ssl_validation",
 88            "certificate",
 89            "skip_path_regex",
 90            "basic_auth_enabled",
 91            "basic_auth_password_attribute",
 92            "basic_auth_user_attribute",
 93            "mode",
 94            "intercept_header_auth",
 95            "redirect_uris",
 96            "cookie_domain",
 97            "jwt_federation_sources",
 98            "jwt_federation_providers",
 99            "access_token_validity",
100            "refresh_token_validity",
101            "outpost_set",
102        ]
103        extra_kwargs = ProviderSerializer.Meta.extra_kwargs
104
105
106class ProxyProviderViewSet(UsedByMixin, ModelViewSet):
107    """ProxyProvider Viewset"""
108
109    queryset = ProxyProvider.objects.all()
110    serializer_class = ProxyProviderSerializer
111    filterset_fields = {
112        "application": ["isnull"],
113        "name": ["iexact"],
114        "authorization_flow__slug": ["iexact"],
115        "property_mappings": ["iexact"],
116        "internal_host": ["iexact"],
117        "external_host": ["iexact"],
118        "internal_host_ssl_validation": ["iexact"],
119        "certificate__kp_uuid": ["iexact"],
120        "certificate__name": ["iexact"],
121        "skip_path_regex": ["iexact"],
122        "basic_auth_enabled": ["iexact"],
123        "basic_auth_password_attribute": ["iexact"],
124        "basic_auth_user_attribute": ["iexact"],
125        "mode": ["iexact"],
126        "cookie_domain": ["iexact"],
127    }
128    search_fields = ["name"]
129    ordering = ["name"]
130
131
132class ProxyOutpostConfigSerializer(ModelSerializer):
133    """Proxy provider serializer for outposts"""
134
135    assigned_application_slug = ReadOnlyField(source="application.slug")
136    assigned_application_name = ReadOnlyField(source="application.name")
137
138    oidc_configuration = SerializerMethodField()
139    access_token_validity = SerializerMethodField()
140    scopes_to_request = SerializerMethodField()
141
142    @extend_schema_field(OpenIDConnectConfigurationSerializer)
143    def get_oidc_configuration(self, obj: ProxyProvider):
144        """Embed OpenID Connect provider information"""
145        return ProviderInfoView(request=self.context["request"]._request).get_info(obj)
146
147    def get_access_token_validity(self, obj: ProxyProvider) -> float | None:
148        """Get token validity as second count"""
149        return timedelta_from_string(obj.access_token_validity).total_seconds()
150
151    def get_scopes_to_request(self, obj: ProxyProvider) -> list[str]:
152        """Get all the scope names the outpost should request,
153        including custom-defined ones"""
154        scope_names = set(
155            ScopeMapping.objects.filter(provider__in=[obj]).values_list("scope_name", flat=True)
156        )
157        return list(scope_names)
158
159    class Meta:
160        model = ProxyProvider
161        fields = [
162            "pk",
163            "name",
164            "internal_host",
165            "external_host",
166            "internal_host_ssl_validation",
167            "client_id",
168            "client_secret",
169            "oidc_configuration",
170            "cookie_secret",
171            "certificate",
172            "skip_path_regex",
173            "basic_auth_enabled",
174            "basic_auth_password_attribute",
175            "basic_auth_user_attribute",
176            "mode",
177            "cookie_domain",
178            "access_token_validity",
179            "intercept_header_auth",
180            "scopes_to_request",
181            "assigned_application_slug",
182            "assigned_application_name",
183        ]
184
185
186class ProxyOutpostConfigViewSet(ListModelMixin, GenericViewSet):
187    """ProxyProvider Viewset"""
188
189    queryset = ProxyProvider.objects.filter(application__isnull=False)
190    serializer_class = ProxyOutpostConfigSerializer
191    ordering = ["name"]
192    search_fields = ["name"]
193    filterset_fields = ["name"]

class OpenIDConnectConfigurationSerializer(authentik.core.api.utils.PassiveSerializer): View Source

23class OpenIDConnectConfigurationSerializer(PassiveSerializer):
24    """rest_framework Serializer for OIDC Configuration"""
25
26    issuer = CharField()
27    authorization_endpoint = CharField()
28    token_endpoint = CharField()
29    userinfo_endpoint = CharField()
30    end_session_endpoint = CharField()
31    introspection_endpoint = CharField()
32    jwks_uri = CharField()
33
34    response_types_supported = ListField(child=CharField())
35    id_token_signing_alg_values_supported = ListField(child=CharField())
36    subject_types_supported = ListField(child=CharField())
37    token_endpoint_auth_methods_supported = ListField(child=CharField())

rest_framework Serializer for OIDC Configuration

issuer

authorization_endpoint

token_endpoint

userinfo_endpoint

end_session_endpoint

introspection_endpoint

jwks_uri

response_types_supported

id_token_signing_alg_values_supported

subject_types_supported

token_endpoint_auth_methods_supported

Inherited Members

authentik.core.api.utils.PassiveSerializer: create; update

class ProxyProviderSerializer(authentik.core.api.providers.ProviderSerializer): View Source

 40class ProxyProviderSerializer(ProviderSerializer):
 41    """ProxyProvider Serializer"""
 42
 43    client_id = CharField(read_only=True)
 44    redirect_uris = RedirectURISerializer(many=True, read_only=True, source="_redirect_uris")
 45    outpost_set = ListField(child=CharField(), read_only=True, source="outpost_set.all")
 46
 47    def validate_basic_auth_enabled(self, value: bool) -> bool:
 48        """Ensure user and password attributes are set"""
 49        if value:
 50            if (
 51                self.initial_data.get("basic_auth_password_attribute", "") == ""
 52                or self.initial_data.get("basic_auth_user_attribute", "") == ""
 53            ):
 54                raise ValidationError(
 55                    _("User and password attributes must be set when basic auth is enabled.")
 56                )
 57        return value
 58
 59    def validate(self, attrs) -> dict[Any, str]:
 60        """Check that internal_host is set when mode is Proxy"""
 61        if (
 62            attrs.get("mode", ProxyMode.PROXY) == ProxyMode.PROXY
 63            and attrs.get("internal_host", "") == ""
 64        ):
 65            raise ValidationError(
 66                {"internal_host": _("Internal host cannot be empty when forward auth is disabled.")}
 67            )
 68        return attrs
 69
 70    def create(self, validated_data: dict):
 71        instance: ProxyProvider = super().create(validated_data)
 72        instance.set_oauth_defaults()
 73        instance.save()
 74        return instance
 75
 76    def update(self, instance: ProxyProvider, validated_data: dict):
 77        instance = super().update(instance, validated_data)
 78        instance.set_oauth_defaults()
 79        instance.save()
 80        return instance
 81
 82    class Meta:
 83        model = ProxyProvider
 84        fields = ProviderSerializer.Meta.fields + [
 85            "client_id",
 86            "internal_host",
 87            "external_host",
 88            "internal_host_ssl_validation",
 89            "certificate",
 90            "skip_path_regex",
 91            "basic_auth_enabled",
 92            "basic_auth_password_attribute",
 93            "basic_auth_user_attribute",
 94            "mode",
 95            "intercept_header_auth",
 96            "redirect_uris",
 97            "cookie_domain",
 98            "jwt_federation_sources",
 99            "jwt_federation_providers",
100            "access_token_validity",
101            "refresh_token_validity",
102            "outpost_set",
103        ]
104        extra_kwargs = ProviderSerializer.Meta.extra_kwargs

ProxyProvider Serializer

client_id

redirect_uris

outpost_set

def validate_basic_auth_enabled(self, value: bool) -> bool: View Source

47    def validate_basic_auth_enabled(self, value: bool) -> bool:
48        """Ensure user and password attributes are set"""
49        if value:
50            if (
51                self.initial_data.get("basic_auth_password_attribute", "") == ""
52                or self.initial_data.get("basic_auth_user_attribute", "") == ""
53            ):
54                raise ValidationError(
55                    _("User and password attributes must be set when basic auth is enabled.")
56                )
57        return value

Ensure user and password attributes are set

def validate(self, attrs) -> dict[typing.Any, str]: View Source

59    def validate(self, attrs) -> dict[Any, str]:
60        """Check that internal_host is set when mode is Proxy"""
61        if (
62            attrs.get("mode", ProxyMode.PROXY) == ProxyMode.PROXY
63            and attrs.get("internal_host", "") == ""
64        ):
65            raise ValidationError(
66                {"internal_host": _("Internal host cannot be empty when forward auth is disabled.")}
67            )
68        return attrs

Check that internal_host is set when mode is Proxy

def create(self, validated_data: dict): View Source

70    def create(self, validated_data: dict):
71        instance: ProxyProvider = super().create(validated_data)
72        instance.set_oauth_defaults()
73        instance.save()
74        return instance

We have a bit of extra checking around this in order to provide descriptive messages when something goes wrong, but this method is essentially just:

return ExampleModel.objects.create(**validated_data)

If there are many to many fields present on the instance then they cannot be set until the model is instantiated, in which case the implementation is like so:

example_relationship = validated_data.pop('example_relationship')
instance = ExampleModel.objects.create(**validated_data)
instance.example_relationship = example_relationship
return instance

The default implementation also does not handle nested relationships. If you want to support writable nested relationships you'll need to write an explicit .create() method.

def update( self, instance: authentik.providers.proxy.models.ProxyProvider, validated_data: dict): View Source

76    def update(self, instance: ProxyProvider, validated_data: dict):
77        instance = super().update(instance, validated_data)
78        instance.set_oauth_defaults()
79        instance.save()
80        return instance

Inherited Members

authentik.core.api.providers.ProviderSerializer: assigned_application_slug; assigned_application_name; assigned_backchannel_application_slug; assigned_backchannel_application_name; component; get_component
authentik.core.api.utils.ModelSerializer: serializer_field_mapping
authentik.core.api.utils.MetaNameSerializer: verbose_name; verbose_name_plural; meta_model_name; get_verbose_name; get_verbose_name_plural; get_meta_model_name

class ProxyProviderSerializer.Meta: View Source

 82    class Meta:
 83        model = ProxyProvider
 84        fields = ProviderSerializer.Meta.fields + [
 85            "client_id",
 86            "internal_host",
 87            "external_host",
 88            "internal_host_ssl_validation",
 89            "certificate",
 90            "skip_path_regex",
 91            "basic_auth_enabled",
 92            "basic_auth_password_attribute",
 93            "basic_auth_user_attribute",
 94            "mode",
 95            "intercept_header_auth",
 96            "redirect_uris",
 97            "cookie_domain",
 98            "jwt_federation_sources",
 99            "jwt_federation_providers",
100            "access_token_validity",
101            "refresh_token_validity",
102            "outpost_set",
103        ]
104        extra_kwargs = ProviderSerializer.Meta.extra_kwargs

model = <class 'authentik.providers.proxy.models.ProxyProvider'>

fields = ['pk', 'name', 'authentication_flow', 'authorization_flow', 'invalidation_flow', 'property_mappings', 'component', 'assigned_application_slug', 'assigned_application_name', 'assigned_backchannel_application_slug', 'assigned_backchannel_application_name', 'verbose_name', 'verbose_name_plural', 'meta_model_name', 'client_id', 'internal_host', 'external_host', 'internal_host_ssl_validation', 'certificate', 'skip_path_regex', 'basic_auth_enabled', 'basic_auth_password_attribute', 'basic_auth_user_attribute', 'mode', 'intercept_header_auth', 'redirect_uris', 'cookie_domain', 'jwt_federation_sources', 'jwt_federation_providers', 'access_token_validity', 'refresh_token_validity', 'outpost_set']

extra_kwargs = {'authorization_flow': {'required': True, 'allow_null': False}, 'invalidation_flow': {'required': True, 'allow_null': False}}

class ProxyProviderViewSet(authentik.core.api.used_by.UsedByMixin, rest_framework.viewsets.ModelViewSet): View Source

107class ProxyProviderViewSet(UsedByMixin, ModelViewSet):
108    """ProxyProvider Viewset"""
109
110    queryset = ProxyProvider.objects.all()
111    serializer_class = ProxyProviderSerializer
112    filterset_fields = {
113        "application": ["isnull"],
114        "name": ["iexact"],
115        "authorization_flow__slug": ["iexact"],
116        "property_mappings": ["iexact"],
117        "internal_host": ["iexact"],
118        "external_host": ["iexact"],
119        "internal_host_ssl_validation": ["iexact"],
120        "certificate__kp_uuid": ["iexact"],
121        "certificate__name": ["iexact"],
122        "skip_path_regex": ["iexact"],
123        "basic_auth_enabled": ["iexact"],
124        "basic_auth_password_attribute": ["iexact"],
125        "basic_auth_user_attribute": ["iexact"],
126        "mode": ["iexact"],
127        "cookie_domain": ["iexact"],
128    }
129    search_fields = ["name"]
130    ordering = ["name"]

ProxyProvider Viewset

queryset = <InheritanceQuerySet []>

serializer_class = <class 'ProxyProviderSerializer'>

filterset_fields = {'application': ['isnull'], 'name': ['iexact'], 'authorization_flow__slug': ['iexact'], 'property_mappings': ['iexact'], 'internal_host': ['iexact'], 'external_host': ['iexact'], 'internal_host_ssl_validation': ['iexact'], 'certificate__kp_uuid': ['iexact'], 'certificate__name': ['iexact'], 'skip_path_regex': ['iexact'], 'basic_auth_enabled': ['iexact'], 'basic_auth_password_attribute': ['iexact'], 'basic_auth_user_attribute': ['iexact'], 'mode': ['iexact'], 'cookie_domain': ['iexact']}

search_fields = ['name']

ordering = ['name']

name = None

description = None

suffix = None

detail = None

basename = None

Inherited Members

authentik.core.api.used_by.UsedByMixin: used_by

class ProxyOutpostConfigSerializer(authentik.core.api.utils.ModelSerializer): View Source

133class ProxyOutpostConfigSerializer(ModelSerializer):
134    """Proxy provider serializer for outposts"""
135
136    assigned_application_slug = ReadOnlyField(source="application.slug")
137    assigned_application_name = ReadOnlyField(source="application.name")
138
139    oidc_configuration = SerializerMethodField()
140    access_token_validity = SerializerMethodField()
141    scopes_to_request = SerializerMethodField()
142
143    @extend_schema_field(OpenIDConnectConfigurationSerializer)
144    def get_oidc_configuration(self, obj: ProxyProvider):
145        """Embed OpenID Connect provider information"""
146        return ProviderInfoView(request=self.context["request"]._request).get_info(obj)
147
148    def get_access_token_validity(self, obj: ProxyProvider) -> float | None:
149        """Get token validity as second count"""
150        return timedelta_from_string(obj.access_token_validity).total_seconds()
151
152    def get_scopes_to_request(self, obj: ProxyProvider) -> list[str]:
153        """Get all the scope names the outpost should request,
154        including custom-defined ones"""
155        scope_names = set(
156            ScopeMapping.objects.filter(provider__in=[obj]).values_list("scope_name", flat=True)
157        )
158        return list(scope_names)
159
160    class Meta:
161        model = ProxyProvider
162        fields = [
163            "pk",
164            "name",
165            "internal_host",
166            "external_host",
167            "internal_host_ssl_validation",
168            "client_id",
169            "client_secret",
170            "oidc_configuration",
171            "cookie_secret",
172            "certificate",
173            "skip_path_regex",
174            "basic_auth_enabled",
175            "basic_auth_password_attribute",
176            "basic_auth_user_attribute",
177            "mode",
178            "cookie_domain",
179            "access_token_validity",
180            "intercept_header_auth",
181            "scopes_to_request",
182            "assigned_application_slug",
183            "assigned_application_name",
184        ]

Proxy provider serializer for outposts

assigned_application_slug

assigned_application_name

oidc_configuration

access_token_validity

scopes_to_request

@extend_schema_field(OpenIDConnectConfigurationSerializer)

def get_oidc_configuration(self, obj: authentik.providers.proxy.models.ProxyProvider): View Source

143    @extend_schema_field(OpenIDConnectConfigurationSerializer)
144    def get_oidc_configuration(self, obj: ProxyProvider):
145        """Embed OpenID Connect provider information"""
146        return ProviderInfoView(request=self.context["request"]._request).get_info(obj)

Embed OpenID Connect provider information

def get_access_token_validity( self, obj: authentik.providers.proxy.models.ProxyProvider) -> float | None: View Source

148    def get_access_token_validity(self, obj: ProxyProvider) -> float | None:
149        """Get token validity as second count"""
150        return timedelta_from_string(obj.access_token_validity).total_seconds()

Get token validity as second count

def get_scopes_to_request(self, obj: authentik.providers.proxy.models.ProxyProvider) -> list[str]: View Source

152    def get_scopes_to_request(self, obj: ProxyProvider) -> list[str]:
153        """Get all the scope names the outpost should request,
154        including custom-defined ones"""
155        scope_names = set(
156            ScopeMapping.objects.filter(provider__in=[obj]).values_list("scope_name", flat=True)
157        )
158        return list(scope_names)

Get all the scope names the outpost should request, including custom-defined ones

Inherited Members

authentik.core.api.utils.ModelSerializer: serializer_field_mapping; update

class ProxyOutpostConfigSerializer.Meta: View Source

160    class Meta:
161        model = ProxyProvider
162        fields = [
163            "pk",
164            "name",
165            "internal_host",
166            "external_host",
167            "internal_host_ssl_validation",
168            "client_id",
169            "client_secret",
170            "oidc_configuration",
171            "cookie_secret",
172            "certificate",
173            "skip_path_regex",
174            "basic_auth_enabled",
175            "basic_auth_password_attribute",
176            "basic_auth_user_attribute",
177            "mode",
178            "cookie_domain",
179            "access_token_validity",
180            "intercept_header_auth",
181            "scopes_to_request",
182            "assigned_application_slug",
183            "assigned_application_name",
184        ]

model = <class 'authentik.providers.proxy.models.ProxyProvider'>

fields = ['pk', 'name', 'internal_host', 'external_host', 'internal_host_ssl_validation', 'client_id', 'client_secret', 'oidc_configuration', 'cookie_secret', 'certificate', 'skip_path_regex', 'basic_auth_enabled', 'basic_auth_password_attribute', 'basic_auth_user_attribute', 'mode', 'cookie_domain', 'access_token_validity', 'intercept_header_auth', 'scopes_to_request', 'assigned_application_slug', 'assigned_application_name']

class ProxyOutpostConfigViewSet(rest_framework.mixins.ListModelMixin, rest_framework.viewsets.GenericViewSet): View Source

187class ProxyOutpostConfigViewSet(ListModelMixin, GenericViewSet):
188    """ProxyProvider Viewset"""
189
190    queryset = ProxyProvider.objects.filter(application__isnull=False)
191    serializer_class = ProxyOutpostConfigSerializer
192    ordering = ["name"]
193    search_fields = ["name"]
194    filterset_fields = ["name"]

ProxyProvider Viewset

queryset = <InheritanceQuerySet []>

serializer_class = <class 'ProxyOutpostConfigSerializer'>

ordering = ['name']

search_fields = ['name']

filterset_fields = ['name']

name = None

description = None

suffix = None

detail = None

basename = None