We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Federation & Social Logins // OAuth - Azure AD // Unable to save settings
**Describe the bug**
See https://github.com/goauthentik/authentik/issues/7744
Problem still exists with release 2023.10.5

**To Reproduce**
Steps to reproduce the behavior:
See https://github.com/goauthentik/authentik/issues/7744

Neither on initial save or update afterwards 'Authorization URL', 'Access token URL ' and 'Profile URL' are changed

**Expected behavior**
A clear and concise description of what you expected to happen.

**Screenshots**
If applicable, add screenshots to help explain your problem.

**Logs**
Seem fine
server:
2023-12-22 11:30:41 {"action": "model_updated", "auth_via": "session", "client_ip": "172.18.0.1", "context": {"http_request": {"args": {}, "method": "PATCH", "path": "/api/v3/sources/oauth/azure-ad/", "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0"}, "model": {"app": "authentik_sources_oauth", "model_name": "oauthsource", "name": "azure ad", "pk": "bda8c9beeac54e5495c8270838e7a797"}}, "event": "Created Event", "host": "localhost:8044", "level": "info", "logger": "authentik.events.models", "pid": 40, "request_id": "ff436debac13400e94274999b5f5d06d", "timestamp": "2023-12-22T10:30:41.555035", "user": {"email": "admin@compose-stack.seven2one.de", "pk": 3, "username": "Admin"}}
2023-12-22 11:30:41 {"auth_via": "session", "event": "Task published", "host": "localhost:8044", "level": "info", "logger": "authentik.root.celery", "pid": 40, "request_id": "ff436debac13400e94274999b5f5d06d", "task_id": "4ebe0d05335f4700b8c0871d6f10235c", "task_name": "authentik.events.tasks.event_notification_handler", "timestamp": "2023-12-22T10:30:41.568890"}
2023-12-22 11:30:41 {"auth_via": "session", "event": "/api/v3/sources/oauth/azure-ad/", "host": "localhost:8044", "level": "info", "logger": "authentik.asgi", "method": "PATCH", "pid": 40, "remote": "172.18.0.1", "request_id": "ff436debac13400e94274999b5f5d06d", "runtime": 396, "scheme": "http", "status": 200, "timestamp": "2023-12-22T10:30:41.572086", "user": "Admin", "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0"}

**Version and Deployment (please complete the following information):**

-   authentik version: 2021.10.5
-   Deployment: docker-compose, helm

**Additional context**



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `authentik/sources/oauth/api/source.py`
Content:
```
1 """OAuth Source Serializer"""
2 from django.urls.base import reverse_lazy
3 from django_filters.filters import BooleanFilter
4 from django_filters.filterset import FilterSet
5 from drf_spectacular.types import OpenApiTypes
6 from drf_spectacular.utils import OpenApiParameter, extend_schema, extend_schema_field
7 from requests import RequestException
8 from rest_framework.decorators import action
9 from rest_framework.fields import BooleanField, CharField, ChoiceField, SerializerMethodField
10 from rest_framework.request import Request
11 from rest_framework.response import Response
12 from rest_framework.serializers import ValidationError
13 from rest_framework.viewsets import ModelViewSet
14 
15 from authentik.core.api.sources import SourceSerializer
16 from authentik.core.api.used_by import UsedByMixin
17 from authentik.core.api.utils import PassiveSerializer
18 from authentik.lib.utils.http import get_http_session
19 from authentik.sources.oauth.models import OAuthSource
20 from authentik.sources.oauth.types.registry import SourceType, registry
21 
22 
23 class SourceTypeSerializer(PassiveSerializer):
24     """Serializer for SourceType"""
25 
26     name = CharField(required=True)
27     verbose_name = CharField(required=True)
28     urls_customizable = BooleanField()
29     request_token_url = CharField(read_only=True, allow_null=True)
30     authorization_url = CharField(read_only=True, allow_null=True)
31     access_token_url = CharField(read_only=True, allow_null=True)
32     profile_url = CharField(read_only=True, allow_null=True)
33     oidc_well_known_url = CharField(read_only=True, allow_null=True)
34     oidc_jwks_url = CharField(read_only=True, allow_null=True)
35 
36 
37 class OAuthSourceSerializer(SourceSerializer):
38     """OAuth Source Serializer"""
39 
40     provider_type = ChoiceField(choices=registry.get_name_tuple())
41     callback_url = SerializerMethodField()
42     type = SerializerMethodField()
43 
44     def get_callback_url(self, instance: OAuthSource) -> str:
45         """Get OAuth Callback URL"""
46         relative_url = reverse_lazy(
47             "authentik_sources_oauth:oauth-client-callback",
48             kwargs={"source_slug": instance.slug},
49         )
50         if "request" not in self.context:
51             return relative_url
52         return self.context["request"].build_absolute_uri(relative_url)
53 
54     @extend_schema_field(SourceTypeSerializer)
55     def get_type(self, instance: OAuthSource) -> SourceTypeSerializer:
56         """Get source's type configuration"""
57         return SourceTypeSerializer(instance.source_type).data
58 
59     def validate(self, attrs: dict) -> dict:
60         session = get_http_session()
61         source_type = registry.find_type(attrs["provider_type"])
62 
63         well_known = attrs.get("oidc_well_known_url") or source_type.oidc_well_known_url
64         inferred_oidc_jwks_url = None
65 
66         if well_known and well_known != "":
67             try:
68                 well_known_config = session.get(well_known)
69                 well_known_config.raise_for_status()
70             except RequestException as exc:
71                 text = exc.response.text if exc.response else str(exc)
72                 raise ValidationError({"oidc_well_known_url": text})
73             config = well_known_config.json()
74             if "issuer" not in config:
75                 raise ValidationError({"oidc_well_known_url": "Invalid well-known configuration"})
76             attrs["authorization_url"] = config.get("authorization_endpoint", "")
77             attrs["access_token_url"] = config.get("token_endpoint", "")
78             attrs["profile_url"] = config.get("userinfo_endpoint", "")
79             inferred_oidc_jwks_url = config.get("jwks_uri", "")
80 
81         # Prefer user-entered URL to inferred URL to default URL
82         jwks_url = attrs.get("oidc_jwks_url") or inferred_oidc_jwks_url or source_type.oidc_jwks_url
83         if jwks_url and jwks_url != "":
84             attrs["oidc_jwks_url"] = jwks_url
85             try:
86                 jwks_config = session.get(jwks_url)
87                 jwks_config.raise_for_status()
88             except RequestException as exc:
89                 text = exc.response.text if exc.response else str(exc)
90                 raise ValidationError({"oidc_jwks_url": text})
91             config = jwks_config.json()
92             attrs["oidc_jwks"] = config
93 
94         provider_type = registry.find_type(attrs.get("provider_type", ""))
95         for url in [
96             "authorization_url",
97             "access_token_url",
98             "profile_url",
99         ]:
100             if getattr(provider_type, url, None) is None:
101                 if url not in attrs:
102                     raise ValidationError(
103                         f"{url} is required for provider {provider_type.verbose_name}"
104                     )
105         return attrs
106 
107     class Meta:
108         model = OAuthSource
109         fields = SourceSerializer.Meta.fields + [
110             "provider_type",
111             "request_token_url",
112             "authorization_url",
113             "access_token_url",
114             "profile_url",
115             "consumer_key",
116             "consumer_secret",
117             "callback_url",
118             "additional_scopes",
119             "type",
120             "oidc_well_known_url",
121             "oidc_jwks_url",
122             "oidc_jwks",
123         ]
124         extra_kwargs = {"consumer_secret": {"write_only": True}}
125 
126 
127 class OAuthSourceFilter(FilterSet):
128     """OAuth Source filter set"""
129 
130     has_jwks = BooleanFilter(label="Only return sources with JWKS data", method="filter_has_jwks")
131 
132     def filter_has_jwks(self, queryset, name, value):  # pragma: no cover
133         """Only return sources with JWKS data"""
134         return queryset.exclude(oidc_jwks__iexact="{}")
135 
136     class Meta:
137         model = OAuthSource
138         fields = [
139             "name",
140             "slug",
141             "enabled",
142             "authentication_flow",
143             "enrollment_flow",
144             "policy_engine_mode",
145             "user_matching_mode",
146             "provider_type",
147             "request_token_url",
148             "authorization_url",
149             "access_token_url",
150             "profile_url",
151             "consumer_key",
152             "additional_scopes",
153         ]
154 
155 
156 class OAuthSourceViewSet(UsedByMixin, ModelViewSet):
157     """Source Viewset"""
158 
159     queryset = OAuthSource.objects.all()
160     serializer_class = OAuthSourceSerializer
161     lookup_field = "slug"
162     filterset_class = OAuthSourceFilter
163     search_fields = ["name", "slug"]
164     ordering = ["name"]
165 
166     @extend_schema(
167         responses={200: SourceTypeSerializer(many=True)},
168         parameters=[
169             OpenApiParameter(
170                 name="name",
171                 location=OpenApiParameter.QUERY,
172                 type=OpenApiTypes.STR,
173             )
174         ],
175     )
176     @action(detail=False, pagination_class=None, filter_backends=[])
177     def source_types(self, request: Request) -> Response:
178         """Get all creatable source types. If ?name is set, only returns the type for <name>.
179         If <name> isn't found, returns the default type."""
180         data = []
181         if "name" in request.query_params:
182             source_type = registry.find_type(request.query_params.get("name"))
183             if source_type.__class__ != SourceType:
184                 data.append(SourceTypeSerializer(source_type).data)
185         else:
186             for source_type in registry.get():
187                 data.append(SourceTypeSerializer(source_type).data)
188         return Response(data)
189
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/authentik/sources/oauth/api/source.py b/authentik/sources/oauth/api/source.py
--- a/authentik/sources/oauth/api/source.py
+++ b/authentik/sources/oauth/api/source.py
@@ -56,6 +56,7 @@
         """Get source's type configuration"""
         return SourceTypeSerializer(instance.source_type).data
 
+    # pylint: disable=too-many-locals
     def validate(self, attrs: dict) -> dict:
         session = get_http_session()
         source_type = registry.find_type(attrs["provider_type"])
@@ -73,9 +74,17 @@
             config = well_known_config.json()
             if "issuer" not in config:
                 raise ValidationError({"oidc_well_known_url": "Invalid well-known configuration"})
-            attrs["authorization_url"] = config.get("authorization_endpoint", "")
-            attrs["access_token_url"] = config.get("token_endpoint", "")
-            attrs["profile_url"] = config.get("userinfo_endpoint", "")
+            field_map = {
+                # authentik field to oidc field
+                "authorization_url": "authorization_endpoint",
+                "access_token_url": "token_endpoint",
+                "profile_url": "userinfo_endpoint",
+            }
+            for ak_key, oidc_key in field_map.items():
+                # Don't overwrite user-set values
+                if ak_key in attrs and attrs[ak_key]:
+                    continue
+                attrs[ak_key] = config.get(oidc_key, "")
             inferred_oidc_jwks_url = config.get("jwks_uri", "")
 
         # Prefer user-entered URL to inferred URL to default URL

{"golden_diff": "diff --git a/authentik/sources/oauth/api/source.py b/authentik/sources/oauth/api/source.py\n--- a/authentik/sources/oauth/api/source.py\n+++ b/authentik/sources/oauth/api/source.py\n@@ -56,6 +56,7 @@\n         \"\"\"Get source's type configuration\"\"\"\n         return SourceTypeSerializer(instance.source_type).data\n \n+    # pylint: disable=too-many-locals\n     def validate(self, attrs: dict) -> dict:\n         session = get_http_session()\n         source_type = registry.find_type(attrs[\"provider_type\"])\n@@ -73,9 +74,17 @@\n             config = well_known_config.json()\n             if \"issuer\" not in config:\n                 raise ValidationError({\"oidc_well_known_url\": \"Invalid well-known configuration\"})\n-            attrs[\"authorization_url\"] = config.get(\"authorization_endpoint\", \"\")\n-            attrs[\"access_token_url\"] = config.get(\"token_endpoint\", \"\")\n-            attrs[\"profile_url\"] = config.get(\"userinfo_endpoint\", \"\")\n+            field_map = {\n+                # authentik field to oidc field\n+                \"authorization_url\": \"authorization_endpoint\",\n+                \"access_token_url\": \"token_endpoint\",\n+                \"profile_url\": \"userinfo_endpoint\",\n+            }\n+            for ak_key, oidc_key in field_map.items():\n+                # Don't overwrite user-set values\n+                if ak_key in attrs and attrs[ak_key]:\n+                    continue\n+                attrs[ak_key] = config.get(oidc_key, \"\")\n             inferred_oidc_jwks_url = config.get(\"jwks_uri\", \"\")\n \n         # Prefer user-entered URL to inferred URL to default URL\n", "issue": "Federation & Social Logins // OAuth - Azure AD // Unable to save settings\n**Describe the bug**\r\nSee https://github.com/goauthentik/authentik/issues/7744\r\nProblem still exists with release 2023.10.5\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior:\r\nSee https://github.com/goauthentik/authentik/issues/7744\r\n\r\nNeither on initial save or update afterwards 'Authorization URL', 'Access token URL ' and 'Profile URL' are changed\r\n\r\n**Expected behavior**\r\nA clear and concise description of what you expected to happen.\r\n\r\n**Screenshots**\r\nIf applicable, add screenshots to help explain your problem.\r\n\r\n**Logs**\r\nSeem fine\r\nserver:\r\n2023-12-22 11:30:41 {\"action\": \"model_updated\", \"auth_via\": \"session\", \"client_ip\": \"172.18.0.1\", \"context\": {\"http_request\": {\"args\": {}, \"method\": \"PATCH\", \"path\": \"/api/v3/sources/oauth/azure-ad/\", \"user_agent\": \"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0\"}, \"model\": {\"app\": \"authentik_sources_oauth\", \"model_name\": \"oauthsource\", \"name\": \"azure ad\", \"pk\": \"bda8c9beeac54e5495c8270838e7a797\"}}, \"event\": \"Created Event\", \"host\": \"localhost:8044\", \"level\": \"info\", \"logger\": \"authentik.events.models\", \"pid\": 40, \"request_id\": \"ff436debac13400e94274999b5f5d06d\", \"timestamp\": \"2023-12-22T10:30:41.555035\", \"user\": {\"email\": \"admin@compose-stack.seven2one.de\", \"pk\": 3, \"username\": \"Admin\"}}\r\n2023-12-22 11:30:41 {\"auth_via\": \"session\", \"event\": \"Task published\", \"host\": \"localhost:8044\", \"level\": \"info\", \"logger\": \"authentik.root.celery\", \"pid\": 40, \"request_id\": \"ff436debac13400e94274999b5f5d06d\", \"task_id\": \"4ebe0d05335f4700b8c0871d6f10235c\", \"task_name\": \"authentik.events.tasks.event_notification_handler\", \"timestamp\": \"2023-12-22T10:30:41.568890\"}\r\n2023-12-22 11:30:41 {\"auth_via\": \"session\", \"event\": \"/api/v3/sources/oauth/azure-ad/\", \"host\": \"localhost:8044\", \"level\": \"info\", \"logger\": \"authentik.asgi\", \"method\": \"PATCH\", \"pid\": 40, \"remote\": \"172.18.0.1\", \"request_id\": \"ff436debac13400e94274999b5f5d06d\", \"runtime\": 396, \"scheme\": \"http\", \"status\": 200, \"timestamp\": \"2023-12-22T10:30:41.572086\", \"user\": \"Admin\", \"user_agent\": \"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0\"}\r\n\r\n**Version and Deployment (please complete the following information):**\r\n\r\n-   authentik version: 2021.10.5\r\n-   Deployment: docker-compose, helm\r\n\r\n**Additional context**\r\n\r\n\n", "before_files": [{"content": "\"\"\"OAuth Source Serializer\"\"\"\nfrom django.urls.base import reverse_lazy\nfrom django_filters.filters import BooleanFilter\nfrom django_filters.filterset import FilterSet\nfrom drf_spectacular.types import OpenApiTypes\nfrom drf_spectacular.utils import OpenApiParameter, extend_schema, extend_schema_field\nfrom requests import RequestException\nfrom rest_framework.decorators import action\nfrom rest_framework.fields import BooleanField, CharField, ChoiceField, SerializerMethodField\nfrom rest_framework.request import Request\nfrom rest_framework.response import Response\nfrom rest_framework.serializers import ValidationError\nfrom rest_framework.viewsets import ModelViewSet\n\nfrom authentik.core.api.sources import SourceSerializer\nfrom authentik.core.api.used_by import UsedByMixin\nfrom authentik.core.api.utils import PassiveSerializer\nfrom authentik.lib.utils.http import get_http_session\nfrom authentik.sources.oauth.models import OAuthSource\nfrom authentik.sources.oauth.types.registry import SourceType, registry\n\n\nclass SourceTypeSerializer(PassiveSerializer):\n    \"\"\"Serializer for SourceType\"\"\"\n\n    name = CharField(required=True)\n    verbose_name = CharField(required=True)\n    urls_customizable = BooleanField()\n    request_token_url = CharField(read_only=True, allow_null=True)\n    authorization_url = CharField(read_only=True, allow_null=True)\n    access_token_url = CharField(read_only=True, allow_null=True)\n    profile_url = CharField(read_only=True, allow_null=True)\n    oidc_well_known_url = CharField(read_only=True, allow_null=True)\n    oidc_jwks_url = CharField(read_only=True, allow_null=True)\n\n\nclass OAuthSourceSerializer(SourceSerializer):\n    \"\"\"OAuth Source Serializer\"\"\"\n\n    provider_type = ChoiceField(choices=registry.get_name_tuple())\n    callback_url = SerializerMethodField()\n    type = SerializerMethodField()\n\n    def get_callback_url(self, instance: OAuthSource) -> str:\n        \"\"\"Get OAuth Callback URL\"\"\"\n        relative_url = reverse_lazy(\n            \"authentik_sources_oauth:oauth-client-callback\",\n            kwargs={\"source_slug\": instance.slug},\n        )\n        if \"request\" not in self.context:\n            return relative_url\n        return self.context[\"request\"].build_absolute_uri(relative_url)\n\n    @extend_schema_field(SourceTypeSerializer)\n    def get_type(self, instance: OAuthSource) -> SourceTypeSerializer:\n        \"\"\"Get source's type configuration\"\"\"\n        return SourceTypeSerializer(instance.source_type).data\n\n    def validate(self, attrs: dict) -> dict:\n        session = get_http_session()\n        source_type = registry.find_type(attrs[\"provider_type\"])\n\n        well_known = attrs.get(\"oidc_well_known_url\") or source_type.oidc_well_known_url\n        inferred_oidc_jwks_url = None\n\n        if well_known and well_known != \"\":\n            try:\n                well_known_config = session.get(well_known)\n                well_known_config.raise_for_status()\n            except RequestException as exc:\n                text = exc.response.text if exc.response else str(exc)\n                raise ValidationError({\"oidc_well_known_url\": text})\n            config = well_known_config.json()\n            if \"issuer\" not in config:\n                raise ValidationError({\"oidc_well_known_url\": \"Invalid well-known configuration\"})\n            attrs[\"authorization_url\"] = config.get(\"authorization_endpoint\", \"\")\n            attrs[\"access_token_url\"] = config.get(\"token_endpoint\", \"\")\n            attrs[\"profile_url\"] = config.get(\"userinfo_endpoint\", \"\")\n            inferred_oidc_jwks_url = config.get(\"jwks_uri\", \"\")\n\n        # Prefer user-entered URL to inferred URL to default URL\n        jwks_url = attrs.get(\"oidc_jwks_url\") or inferred_oidc_jwks_url or source_type.oidc_jwks_url\n        if jwks_url and jwks_url != \"\":\n            attrs[\"oidc_jwks_url\"] = jwks_url\n            try:\n                jwks_config = session.get(jwks_url)\n                jwks_config.raise_for_status()\n            except RequestException as exc:\n                text = exc.response.text if exc.response else str(exc)\n                raise ValidationError({\"oidc_jwks_url\": text})\n            config = jwks_config.json()\n            attrs[\"oidc_jwks\"] = config\n\n        provider_type = registry.find_type(attrs.get(\"provider_type\", \"\"))\n        for url in [\n            \"authorization_url\",\n            \"access_token_url\",\n            \"profile_url\",\n        ]:\n            if getattr(provider_type, url, None) is None:\n                if url not in attrs:\n                    raise ValidationError(\n                        f\"{url} is required for provider {provider_type.verbose_name}\"\n                    )\n        return attrs\n\n    class Meta:\n        model = OAuthSource\n        fields = SourceSerializer.Meta.fields + [\n            \"provider_type\",\n            \"request_token_url\",\n            \"authorization_url\",\n            \"access_token_url\",\n            \"profile_url\",\n            \"consumer_key\",\n            \"consumer_secret\",\n            \"callback_url\",\n            \"additional_scopes\",\n            \"type\",\n            \"oidc_well_known_url\",\n            \"oidc_jwks_url\",\n            \"oidc_jwks\",\n        ]\n        extra_kwargs = {\"consumer_secret\": {\"write_only\": True}}\n\n\nclass OAuthSourceFilter(FilterSet):\n    \"\"\"OAuth Source filter set\"\"\"\n\n    has_jwks = BooleanFilter(label=\"Only return sources with JWKS data\", method=\"filter_has_jwks\")\n\n    def filter_has_jwks(self, queryset, name, value):  # pragma: no cover\n        \"\"\"Only return sources with JWKS data\"\"\"\n        return queryset.exclude(oidc_jwks__iexact=\"{}\")\n\n    class Meta:\n        model = OAuthSource\n        fields = [\n            \"name\",\n            \"slug\",\n            \"enabled\",\n            \"authentication_flow\",\n            \"enrollment_flow\",\n            \"policy_engine_mode\",\n            \"user_matching_mode\",\n            \"provider_type\",\n            \"request_token_url\",\n            \"authorization_url\",\n            \"access_token_url\",\n            \"profile_url\",\n            \"consumer_key\",\n            \"additional_scopes\",\n        ]\n\n\nclass OAuthSourceViewSet(UsedByMixin, ModelViewSet):\n    \"\"\"Source Viewset\"\"\"\n\n    queryset = OAuthSource.objects.all()\n    serializer_class = OAuthSourceSerializer\n    lookup_field = \"slug\"\n    filterset_class = OAuthSourceFilter\n    search_fields = [\"name\", \"slug\"]\n    ordering = [\"name\"]\n\n    @extend_schema(\n        responses={200: SourceTypeSerializer(many=True)},\n        parameters=[\n            OpenApiParameter(\n                name=\"name\",\n                location=OpenApiParameter.QUERY,\n                type=OpenApiTypes.STR,\n            )\n        ],\n    )\n    @action(detail=False, pagination_class=None, filter_backends=[])\n    def source_types(self, request: Request) -> Response:\n        \"\"\"Get all creatable source types. If ?name is set, only returns the type for <name>.\n        If <name> isn't found, returns the default type.\"\"\"\n        data = []\n        if \"name\" in request.query_params:\n            source_type = registry.find_type(request.query_params.get(\"name\"))\n            if source_type.__class__ != SourceType:\n                data.append(SourceTypeSerializer(source_type).data)\n        else:\n            for source_type in registry.get():\n                data.append(SourceTypeSerializer(source_type).data)\n        return Response(data)\n", "path": "authentik/sources/oauth/api/source.py"}], "after_files": [{"content": "\"\"\"OAuth Source Serializer\"\"\"\nfrom django.urls.base import reverse_lazy\nfrom django_filters.filters import BooleanFilter\nfrom django_filters.filterset import FilterSet\nfrom drf_spectacular.types import OpenApiTypes\nfrom drf_spectacular.utils import OpenApiParameter, extend_schema, extend_schema_field\nfrom requests import RequestException\nfrom rest_framework.decorators import action\nfrom rest_framework.fields import BooleanField, CharField, ChoiceField, SerializerMethodField\nfrom rest_framework.request import Request\nfrom rest_framework.response import Response\nfrom rest_framework.serializers import ValidationError\nfrom rest_framework.viewsets import ModelViewSet\n\nfrom authentik.core.api.sources import SourceSerializer\nfrom authentik.core.api.used_by import UsedByMixin\nfrom authentik.core.api.utils import PassiveSerializer\nfrom authentik.lib.utils.http import get_http_session\nfrom authentik.sources.oauth.models import OAuthSource\nfrom authentik.sources.oauth.types.registry import SourceType, registry\n\n\nclass SourceTypeSerializer(PassiveSerializer):\n    \"\"\"Serializer for SourceType\"\"\"\n\n    name = CharField(required=True)\n    verbose_name = CharField(required=True)\n    urls_customizable = BooleanField()\n    request_token_url = CharField(read_only=True, allow_null=True)\n    authorization_url = CharField(read_only=True, allow_null=True)\n    access_token_url = CharField(read_only=True, allow_null=True)\n    profile_url = CharField(read_only=True, allow_null=True)\n    oidc_well_known_url = CharField(read_only=True, allow_null=True)\n    oidc_jwks_url = CharField(read_only=True, allow_null=True)\n\n\nclass OAuthSourceSerializer(SourceSerializer):\n    \"\"\"OAuth Source Serializer\"\"\"\n\n    provider_type = ChoiceField(choices=registry.get_name_tuple())\n    callback_url = SerializerMethodField()\n    type = SerializerMethodField()\n\n    def get_callback_url(self, instance: OAuthSource) -> str:\n        \"\"\"Get OAuth Callback URL\"\"\"\n        relative_url = reverse_lazy(\n            \"authentik_sources_oauth:oauth-client-callback\",\n            kwargs={\"source_slug\": instance.slug},\n        )\n        if \"request\" not in self.context:\n            return relative_url\n        return self.context[\"request\"].build_absolute_uri(relative_url)\n\n    @extend_schema_field(SourceTypeSerializer)\n    def get_type(self, instance: OAuthSource) -> SourceTypeSerializer:\n        \"\"\"Get source's type configuration\"\"\"\n        return SourceTypeSerializer(instance.source_type).data\n\n    # pylint: disable=too-many-locals\n    def validate(self, attrs: dict) -> dict:\n        session = get_http_session()\n        source_type = registry.find_type(attrs[\"provider_type\"])\n\n        well_known = attrs.get(\"oidc_well_known_url\") or source_type.oidc_well_known_url\n        inferred_oidc_jwks_url = None\n\n        if well_known and well_known != \"\":\n            try:\n                well_known_config = session.get(well_known)\n                well_known_config.raise_for_status()\n            except RequestException as exc:\n                text = exc.response.text if exc.response else str(exc)\n                raise ValidationError({\"oidc_well_known_url\": text})\n            config = well_known_config.json()\n            if \"issuer\" not in config:\n                raise ValidationError({\"oidc_well_known_url\": \"Invalid well-known configuration\"})\n            field_map = {\n                # authentik field to oidc field\n                \"authorization_url\": \"authorization_endpoint\",\n                \"access_token_url\": \"token_endpoint\",\n                \"profile_url\": \"userinfo_endpoint\",\n            }\n            for ak_key, oidc_key in field_map.items():\n                # Don't overwrite user-set values\n                if ak_key in attrs and attrs[ak_key]:\n                    continue\n                attrs[ak_key] = config.get(oidc_key, \"\")\n            inferred_oidc_jwks_url = config.get(\"jwks_uri\", \"\")\n\n        # Prefer user-entered URL to inferred URL to default URL\n        jwks_url = attrs.get(\"oidc_jwks_url\") or inferred_oidc_jwks_url or source_type.oidc_jwks_url\n        if jwks_url and jwks_url != \"\":\n            attrs[\"oidc_jwks_url\"] = jwks_url\n            try:\n                jwks_config = session.get(jwks_url)\n                jwks_config.raise_for_status()\n            except RequestException as exc:\n                text = exc.response.text if exc.response else str(exc)\n                raise ValidationError({\"oidc_jwks_url\": text})\n            config = jwks_config.json()\n            attrs[\"oidc_jwks\"] = config\n\n        provider_type = registry.find_type(attrs.get(\"provider_type\", \"\"))\n        for url in [\n            \"authorization_url\",\n            \"access_token_url\",\n            \"profile_url\",\n        ]:\n            if getattr(provider_type, url, None) is None:\n                if url not in attrs:\n                    raise ValidationError(\n                        f\"{url} is required for provider {provider_type.verbose_name}\"\n                    )\n        return attrs\n\n    class Meta:\n        model = OAuthSource\n        fields = SourceSerializer.Meta.fields + [\n            \"provider_type\",\n            \"request_token_url\",\n            \"authorization_url\",\n            \"access_token_url\",\n            \"profile_url\",\n            \"consumer_key\",\n            \"consumer_secret\",\n            \"callback_url\",\n            \"additional_scopes\",\n            \"type\",\n            \"oidc_well_known_url\",\n            \"oidc_jwks_url\",\n            \"oidc_jwks\",\n        ]\n        extra_kwargs = {\"consumer_secret\": {\"write_only\": True}}\n\n\nclass OAuthSourceFilter(FilterSet):\n    \"\"\"OAuth Source filter set\"\"\"\n\n    has_jwks = BooleanFilter(label=\"Only return sources with JWKS data\", method=\"filter_has_jwks\")\n\n    def filter_has_jwks(self, queryset, name, value):  # pragma: no cover\n        \"\"\"Only return sources with JWKS data\"\"\"\n        return queryset.exclude(oidc_jwks__iexact=\"{}\")\n\n    class Meta:\n        model = OAuthSource\n        fields = [\n            \"name\",\n            \"slug\",\n            \"enabled\",\n            \"authentication_flow\",\n            \"enrollment_flow\",\n            \"policy_engine_mode\",\n            \"user_matching_mode\",\n            \"provider_type\",\n            \"request_token_url\",\n            \"authorization_url\",\n            \"access_token_url\",\n            \"profile_url\",\n            \"consumer_key\",\n            \"additional_scopes\",\n        ]\n\n\nclass OAuthSourceViewSet(UsedByMixin, ModelViewSet):\n    \"\"\"Source Viewset\"\"\"\n\n    queryset = OAuthSource.objects.all()\n    serializer_class = OAuthSourceSerializer\n    lookup_field = \"slug\"\n    filterset_class = OAuthSourceFilter\n    search_fields = [\"name\", \"slug\"]\n    ordering = [\"name\"]\n\n    @extend_schema(\n        responses={200: SourceTypeSerializer(many=True)},\n        parameters=[\n            OpenApiParameter(\n                name=\"name\",\n                location=OpenApiParameter.QUERY,\n                type=OpenApiTypes.STR,\n            )\n        ],\n    )\n    @action(detail=False, pagination_class=None, filter_backends=[])\n    def source_types(self, request: Request) -> Response:\n        \"\"\"Get all creatable source types. If ?name is set, only returns the type for <name>.\n        If <name> isn't found, returns the default type.\"\"\"\n        data = []\n        if \"name\" in request.query_params:\n            source_type = registry.find_type(request.query_params.get(\"name\"))\n            if source_type.__class__ != SourceType:\n                data.append(SourceTypeSerializer(source_type).data)\n        else:\n            for source_type in registry.get():\n                data.append(SourceTypeSerializer(source_type).data)\n        return Response(data)\n", "path": "authentik/sources/oauth/api/source.py"}]}