We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
http-basic-auth headers for external OAuth source
**Describe your question**
 Is there a way to have authentik send http-basic-auth headers (as per [RFC6749 2.3.1](https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1)) for external IdPs? I can't find any settings regarding this and it doesn't seem to do that by default.

**Relevant info**
An external IdP we use as a login source updated their underlying software. They now require applications to send http-basic-auth headers for OIDC client authentication. Before the update, login worked just fine.
Now we receive an error message: "Authentication failed: Could not retrieve token." and the logs show an HTTP 401 Unauthorized error, when trying to reach the IdP's token endpoint.

**Logs**
server-1:
{
    "auth_via": "unauthenticated",
    "domain_url": "[authentik.idp]", 
    "event": "Unable to fetch access token", 
    "exc": "HTTPError('401 Client Error: Unauthorized for url: https://[external.idp]/oauth2/token')",
    "host": "[authentik.idp]",
    "level": "warning",
    "logger": "authentik.sources.oauth.clients.oauth2",
    "pid": 55,
    "request_id": "51bca021eac7412bb2e54233753761cf",
    "response": "401 Client Error: Unauthorized for url: https://[external.idp]/oauth2/token",
    "schema_name": "public",
    "timestamp": "2024-04-15T11:22:40.705924"
}

Note that [url.idp] is redacted.

**Version and Deployment:**

-   authentik version: 2024.2.2
-   Deployment: docker-compose

--- 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/clients/oauth2.py`
Content:
```
1 """OAuth 2 Clients"""
2 
3 from json import loads
4 from typing import Any
5 from urllib.parse import parse_qsl
6 
7 from django.utils.crypto import constant_time_compare, get_random_string
8 from django.utils.translation import gettext as _
9 from requests.exceptions import RequestException
10 from requests.models import Response
11 from structlog.stdlib import get_logger
12 
13 from authentik.sources.oauth.clients.base import BaseOAuthClient
14 
15 LOGGER = get_logger()
16 SESSION_KEY_OAUTH_PKCE = "authentik/sources/oauth/pkce"
17 
18 
19 class OAuth2Client(BaseOAuthClient):
20     """OAuth2 Client"""
21 
22     _default_headers = {
23         "Accept": "application/json",
24     }
25 
26     def get_request_arg(self, key: str, default: Any | None = None) -> Any:
27         """Depending on request type, get data from post or get"""
28         if self.request.method == "POST":
29             return self.request.POST.get(key, default)
30         return self.request.GET.get(key, default)
31 
32     def check_application_state(self) -> bool:
33         """Check optional state parameter."""
34         stored = self.request.session.get(self.session_key, None)
35         returned = self.get_request_arg("state", None)
36         check = False
37         if stored is not None:
38             if returned is not None:
39                 check = constant_time_compare(stored, returned)
40             else:
41                 LOGGER.warning("No state parameter returned by the source.")
42         else:
43             LOGGER.warning("No state stored in the session.")
44         return check
45 
46     def get_application_state(self) -> str:
47         """Generate state optional parameter."""
48         return get_random_string(32)
49 
50     def get_client_id(self) -> str:
51         """Get client id"""
52         return self.source.consumer_key
53 
54     def get_client_secret(self) -> str:
55         """Get client secret"""
56         return self.source.consumer_secret
57 
58     def get_access_token(self, **request_kwargs) -> dict[str, Any] | None:
59         """Fetch access token from callback request."""
60         callback = self.request.build_absolute_uri(self.callback or self.request.path)
61         if not self.check_application_state():
62             LOGGER.warning("Application state check failed.")
63             return {"error": "State check failed."}
64         code = self.get_request_arg("code", None)
65         if not code:
66             LOGGER.warning("No code returned by the source")
67             error = self.get_request_arg("error", None)
68             error_desc = self.get_request_arg("error_description", None)
69             return {"error": error_desc or error or _("No token received.")}
70         args = {
71             "client_id": self.get_client_id(),
72             "client_secret": self.get_client_secret(),
73             "redirect_uri": callback,
74             "code": code,
75             "grant_type": "authorization_code",
76         }
77         if SESSION_KEY_OAUTH_PKCE in self.request.session:
78             args["code_verifier"] = self.request.session[SESSION_KEY_OAUTH_PKCE]
79         try:
80             access_token_url = self.source.source_type.access_token_url or ""
81             if self.source.source_type.urls_customizable and self.source.access_token_url:
82                 access_token_url = self.source.access_token_url
83             response = self.session.request(
84                 "post", access_token_url, data=args, headers=self._default_headers, **request_kwargs
85             )
86             response.raise_for_status()
87         except RequestException as exc:
88             LOGGER.warning(
89                 "Unable to fetch access token",
90                 exc=exc,
91                 response=exc.response.text if exc.response else str(exc),
92             )
93             return None
94         return response.json()
95 
96     def get_redirect_args(self) -> dict[str, str]:
97         """Get request parameters for redirect url."""
98         callback = self.request.build_absolute_uri(self.callback)
99         client_id: str = self.get_client_id()
100         args: dict[str, str] = {
101             "client_id": client_id,
102             "redirect_uri": callback,
103             "response_type": "code",
104         }
105         state = self.get_application_state()
106         if state is not None:
107             args["state"] = state
108             self.request.session[self.session_key] = state
109         return args
110 
111     def parse_raw_token(self, raw_token: str) -> dict[str, Any]:
112         """Parse token and secret from raw token response."""
113         # Load as json first then parse as query string
114         try:
115             token_data = loads(raw_token)
116         except ValueError:
117             return dict(parse_qsl(raw_token))
118         return token_data
119 
120     def do_request(self, method: str, url: str, **kwargs) -> Response:
121         """Build remote url request. Constructs necessary auth."""
122         if "token" in kwargs:
123             token = kwargs.pop("token")
124 
125             params = kwargs.get("params", {})
126             params["access_token"] = token["access_token"]
127             kwargs["params"] = params
128 
129             headers = kwargs.get("headers", {})
130             headers["Authorization"] = f"{token['token_type']} {token['access_token']}"
131             kwargs["headers"] = headers
132         return super().do_request(method, url, **kwargs)
133 
134     @property
135     def session_key(self):
136         return f"oauth-client-{self.source.name}-request-state"
137 
138 
139 class UserprofileHeaderAuthClient(OAuth2Client):
140     """OAuth client which only sends authentication via header, not querystring"""
141 
142     def get_profile_info(self, token: dict[str, str]) -> dict[str, Any] | None:
143         "Fetch user profile information."
144         profile_url = self.source.source_type.profile_url or ""
145         if self.source.source_type.urls_customizable and self.source.profile_url:
146             profile_url = self.source.profile_url
147         response = self.session.request(
148             "get",
149             profile_url,
150             headers={"Authorization": f"{token['token_type']} {token['access_token']}"},
151         )
152         try:
153             response.raise_for_status()
154         except RequestException as exc:
155             LOGGER.warning(
156                 "Unable to fetch user profile",
157                 exc=exc,
158                 response=exc.response.text if exc.response else str(exc),
159             )
160             return None
161         return response.json()
162
```
--- 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/clients/oauth2.py b/authentik/sources/oauth/clients/oauth2.py
--- a/authentik/sources/oauth/clients/oauth2.py
+++ b/authentik/sources/oauth/clients/oauth2.py
@@ -80,7 +80,7 @@
             access_token_url = self.source.source_type.access_token_url or ""
             if self.source.source_type.urls_customizable and self.source.access_token_url:
                 access_token_url = self.source.access_token_url
-            response = self.session.request(
+            response = self.do_request(
                 "post", access_token_url, data=args, headers=self._default_headers, **request_kwargs
             )
             response.raise_for_status()

{"golden_diff": "diff --git a/authentik/sources/oauth/clients/oauth2.py b/authentik/sources/oauth/clients/oauth2.py\n--- a/authentik/sources/oauth/clients/oauth2.py\n+++ b/authentik/sources/oauth/clients/oauth2.py\n@@ -80,7 +80,7 @@\n             access_token_url = self.source.source_type.access_token_url or \"\"\n             if self.source.source_type.urls_customizable and self.source.access_token_url:\n                 access_token_url = self.source.access_token_url\n-            response = self.session.request(\n+            response = self.do_request(\n                 \"post\", access_token_url, data=args, headers=self._default_headers, **request_kwargs\n             )\n             response.raise_for_status()\n", "issue": "http-basic-auth headers for external OAuth source\n**Describe your question**\r\n Is there a way to have authentik send http-basic-auth headers (as per [RFC6749 2.3.1](https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1)) for external IdPs? I can't find any settings regarding this and it doesn't seem to do that by default.\r\n\r\n**Relevant info**\r\nAn external IdP we use as a login source updated their underlying software. They now require applications to send http-basic-auth headers for OIDC client authentication. Before the update, login worked just fine.\r\nNow we receive an error message: \"Authentication failed: Could not retrieve token.\" and the logs show an HTTP 401 Unauthorized error, when trying to reach the IdP's token endpoint.\r\n\r\n**Logs**\r\nserver-1:\r\n{\r\n    \"auth_via\": \"unauthenticated\",\r\n    \"domain_url\": \"[authentik.idp]\", \r\n    \"event\": \"Unable to fetch access token\", \r\n    \"exc\": \"HTTPError('401 Client Error: Unauthorized for url: https://[external.idp]/oauth2/token')\",\r\n    \"host\": \"[authentik.idp]\",\r\n    \"level\": \"warning\",\r\n    \"logger\": \"authentik.sources.oauth.clients.oauth2\",\r\n    \"pid\": 55,\r\n    \"request_id\": \"51bca021eac7412bb2e54233753761cf\",\r\n    \"response\": \"401 Client Error: Unauthorized for url: https://[external.idp]/oauth2/token\",\r\n    \"schema_name\": \"public\",\r\n    \"timestamp\": \"2024-04-15T11:22:40.705924\"\r\n}\r\n\r\nNote that [url.idp] is redacted.\r\n\r\n**Version and Deployment:**\r\n\r\n-   authentik version: 2024.2.2\r\n-   Deployment: docker-compose\n", "before_files": [{"content": "\"\"\"OAuth 2 Clients\"\"\"\n\nfrom json import loads\nfrom typing import Any\nfrom urllib.parse import parse_qsl\n\nfrom django.utils.crypto import constant_time_compare, get_random_string\nfrom django.utils.translation import gettext as _\nfrom requests.exceptions import RequestException\nfrom requests.models import Response\nfrom structlog.stdlib import get_logger\n\nfrom authentik.sources.oauth.clients.base import BaseOAuthClient\n\nLOGGER = get_logger()\nSESSION_KEY_OAUTH_PKCE = \"authentik/sources/oauth/pkce\"\n\n\nclass OAuth2Client(BaseOAuthClient):\n    \"\"\"OAuth2 Client\"\"\"\n\n    _default_headers = {\n        \"Accept\": \"application/json\",\n    }\n\n    def get_request_arg(self, key: str, default: Any | None = None) -> Any:\n        \"\"\"Depending on request type, get data from post or get\"\"\"\n        if self.request.method == \"POST\":\n            return self.request.POST.get(key, default)\n        return self.request.GET.get(key, default)\n\n    def check_application_state(self) -> bool:\n        \"\"\"Check optional state parameter.\"\"\"\n        stored = self.request.session.get(self.session_key, None)\n        returned = self.get_request_arg(\"state\", None)\n        check = False\n        if stored is not None:\n            if returned is not None:\n                check = constant_time_compare(stored, returned)\n            else:\n                LOGGER.warning(\"No state parameter returned by the source.\")\n        else:\n            LOGGER.warning(\"No state stored in the session.\")\n        return check\n\n    def get_application_state(self) -> str:\n        \"\"\"Generate state optional parameter.\"\"\"\n        return get_random_string(32)\n\n    def get_client_id(self) -> str:\n        \"\"\"Get client id\"\"\"\n        return self.source.consumer_key\n\n    def get_client_secret(self) -> str:\n        \"\"\"Get client secret\"\"\"\n        return self.source.consumer_secret\n\n    def get_access_token(self, **request_kwargs) -> dict[str, Any] | None:\n        \"\"\"Fetch access token from callback request.\"\"\"\n        callback = self.request.build_absolute_uri(self.callback or self.request.path)\n        if not self.check_application_state():\n            LOGGER.warning(\"Application state check failed.\")\n            return {\"error\": \"State check failed.\"}\n        code = self.get_request_arg(\"code\", None)\n        if not code:\n            LOGGER.warning(\"No code returned by the source\")\n            error = self.get_request_arg(\"error\", None)\n            error_desc = self.get_request_arg(\"error_description\", None)\n            return {\"error\": error_desc or error or _(\"No token received.\")}\n        args = {\n            \"client_id\": self.get_client_id(),\n            \"client_secret\": self.get_client_secret(),\n            \"redirect_uri\": callback,\n            \"code\": code,\n            \"grant_type\": \"authorization_code\",\n        }\n        if SESSION_KEY_OAUTH_PKCE in self.request.session:\n            args[\"code_verifier\"] = self.request.session[SESSION_KEY_OAUTH_PKCE]\n        try:\n            access_token_url = self.source.source_type.access_token_url or \"\"\n            if self.source.source_type.urls_customizable and self.source.access_token_url:\n                access_token_url = self.source.access_token_url\n            response = self.session.request(\n                \"post\", access_token_url, data=args, headers=self._default_headers, **request_kwargs\n            )\n            response.raise_for_status()\n        except RequestException as exc:\n            LOGGER.warning(\n                \"Unable to fetch access token\",\n                exc=exc,\n                response=exc.response.text if exc.response else str(exc),\n            )\n            return None\n        return response.json()\n\n    def get_redirect_args(self) -> dict[str, str]:\n        \"\"\"Get request parameters for redirect url.\"\"\"\n        callback = self.request.build_absolute_uri(self.callback)\n        client_id: str = self.get_client_id()\n        args: dict[str, str] = {\n            \"client_id\": client_id,\n            \"redirect_uri\": callback,\n            \"response_type\": \"code\",\n        }\n        state = self.get_application_state()\n        if state is not None:\n            args[\"state\"] = state\n            self.request.session[self.session_key] = state\n        return args\n\n    def parse_raw_token(self, raw_token: str) -> dict[str, Any]:\n        \"\"\"Parse token and secret from raw token response.\"\"\"\n        # Load as json first then parse as query string\n        try:\n            token_data = loads(raw_token)\n        except ValueError:\n            return dict(parse_qsl(raw_token))\n        return token_data\n\n    def do_request(self, method: str, url: str, **kwargs) -> Response:\n        \"\"\"Build remote url request. Constructs necessary auth.\"\"\"\n        if \"token\" in kwargs:\n            token = kwargs.pop(\"token\")\n\n            params = kwargs.get(\"params\", {})\n            params[\"access_token\"] = token[\"access_token\"]\n            kwargs[\"params\"] = params\n\n            headers = kwargs.get(\"headers\", {})\n            headers[\"Authorization\"] = f\"{token['token_type']} {token['access_token']}\"\n            kwargs[\"headers\"] = headers\n        return super().do_request(method, url, **kwargs)\n\n    @property\n    def session_key(self):\n        return f\"oauth-client-{self.source.name}-request-state\"\n\n\nclass UserprofileHeaderAuthClient(OAuth2Client):\n    \"\"\"OAuth client which only sends authentication via header, not querystring\"\"\"\n\n    def get_profile_info(self, token: dict[str, str]) -> dict[str, Any] | None:\n        \"Fetch user profile information.\"\n        profile_url = self.source.source_type.profile_url or \"\"\n        if self.source.source_type.urls_customizable and self.source.profile_url:\n            profile_url = self.source.profile_url\n        response = self.session.request(\n            \"get\",\n            profile_url,\n            headers={\"Authorization\": f\"{token['token_type']} {token['access_token']}\"},\n        )\n        try:\n            response.raise_for_status()\n        except RequestException as exc:\n            LOGGER.warning(\n                \"Unable to fetch user profile\",\n                exc=exc,\n                response=exc.response.text if exc.response else str(exc),\n            )\n            return None\n        return response.json()\n", "path": "authentik/sources/oauth/clients/oauth2.py"}], "after_files": [{"content": "\"\"\"OAuth 2 Clients\"\"\"\n\nfrom json import loads\nfrom typing import Any\nfrom urllib.parse import parse_qsl\n\nfrom django.utils.crypto import constant_time_compare, get_random_string\nfrom django.utils.translation import gettext as _\nfrom requests.exceptions import RequestException\nfrom requests.models import Response\nfrom structlog.stdlib import get_logger\n\nfrom authentik.sources.oauth.clients.base import BaseOAuthClient\n\nLOGGER = get_logger()\nSESSION_KEY_OAUTH_PKCE = \"authentik/sources/oauth/pkce\"\n\n\nclass OAuth2Client(BaseOAuthClient):\n    \"\"\"OAuth2 Client\"\"\"\n\n    _default_headers = {\n        \"Accept\": \"application/json\",\n    }\n\n    def get_request_arg(self, key: str, default: Any | None = None) -> Any:\n        \"\"\"Depending on request type, get data from post or get\"\"\"\n        if self.request.method == \"POST\":\n            return self.request.POST.get(key, default)\n        return self.request.GET.get(key, default)\n\n    def check_application_state(self) -> bool:\n        \"\"\"Check optional state parameter.\"\"\"\n        stored = self.request.session.get(self.session_key, None)\n        returned = self.get_request_arg(\"state\", None)\n        check = False\n        if stored is not None:\n            if returned is not None:\n                check = constant_time_compare(stored, returned)\n            else:\n                LOGGER.warning(\"No state parameter returned by the source.\")\n        else:\n            LOGGER.warning(\"No state stored in the session.\")\n        return check\n\n    def get_application_state(self) -> str:\n        \"\"\"Generate state optional parameter.\"\"\"\n        return get_random_string(32)\n\n    def get_client_id(self) -> str:\n        \"\"\"Get client id\"\"\"\n        return self.source.consumer_key\n\n    def get_client_secret(self) -> str:\n        \"\"\"Get client secret\"\"\"\n        return self.source.consumer_secret\n\n    def get_access_token(self, **request_kwargs) -> dict[str, Any] | None:\n        \"\"\"Fetch access token from callback request.\"\"\"\n        callback = self.request.build_absolute_uri(self.callback or self.request.path)\n        if not self.check_application_state():\n            LOGGER.warning(\"Application state check failed.\")\n            return {\"error\": \"State check failed.\"}\n        code = self.get_request_arg(\"code\", None)\n        if not code:\n            LOGGER.warning(\"No code returned by the source\")\n            error = self.get_request_arg(\"error\", None)\n            error_desc = self.get_request_arg(\"error_description\", None)\n            return {\"error\": error_desc or error or _(\"No token received.\")}\n        args = {\n            \"client_id\": self.get_client_id(),\n            \"client_secret\": self.get_client_secret(),\n            \"redirect_uri\": callback,\n            \"code\": code,\n            \"grant_type\": \"authorization_code\",\n        }\n        if SESSION_KEY_OAUTH_PKCE in self.request.session:\n            args[\"code_verifier\"] = self.request.session[SESSION_KEY_OAUTH_PKCE]\n        try:\n            access_token_url = self.source.source_type.access_token_url or \"\"\n            if self.source.source_type.urls_customizable and self.source.access_token_url:\n                access_token_url = self.source.access_token_url\n            response = self.do_request(\n                \"post\", access_token_url, data=args, headers=self._default_headers, **request_kwargs\n            )\n            response.raise_for_status()\n        except RequestException as exc:\n            LOGGER.warning(\n                \"Unable to fetch access token\",\n                exc=exc,\n                response=exc.response.text if exc.response else str(exc),\n            )\n            return None\n        return response.json()\n\n    def get_redirect_args(self) -> dict[str, str]:\n        \"\"\"Get request parameters for redirect url.\"\"\"\n        callback = self.request.build_absolute_uri(self.callback)\n        client_id: str = self.get_client_id()\n        args: dict[str, str] = {\n            \"client_id\": client_id,\n            \"redirect_uri\": callback,\n            \"response_type\": \"code\",\n        }\n        state = self.get_application_state()\n        if state is not None:\n            args[\"state\"] = state\n            self.request.session[self.session_key] = state\n        return args\n\n    def parse_raw_token(self, raw_token: str) -> dict[str, Any]:\n        \"\"\"Parse token and secret from raw token response.\"\"\"\n        # Load as json first then parse as query string\n        try:\n            token_data = loads(raw_token)\n        except ValueError:\n            return dict(parse_qsl(raw_token))\n        return token_data\n\n    def do_request(self, method: str, url: str, **kwargs) -> Response:\n        \"\"\"Build remote url request. Constructs necessary auth.\"\"\"\n        if \"token\" in kwargs:\n            token = kwargs.pop(\"token\")\n\n            params = kwargs.get(\"params\", {})\n            params[\"access_token\"] = token[\"access_token\"]\n            kwargs[\"params\"] = params\n\n            headers = kwargs.get(\"headers\", {})\n            headers[\"Authorization\"] = f\"{token['token_type']} {token['access_token']}\"\n            kwargs[\"headers\"] = headers\n        return super().do_request(method, url, **kwargs)\n\n    @property\n    def session_key(self):\n        return f\"oauth-client-{self.source.name}-request-state\"\n\n\nclass UserprofileHeaderAuthClient(OAuth2Client):\n    \"\"\"OAuth client which only sends authentication via header, not querystring\"\"\"\n\n    def get_profile_info(self, token: dict[str, str]) -> dict[str, Any] | None:\n        \"Fetch user profile information.\"\n        profile_url = self.source.source_type.profile_url or \"\"\n        if self.source.source_type.urls_customizable and self.source.profile_url:\n            profile_url = self.source.profile_url\n        response = self.session.request(\n            \"get\",\n            profile_url,\n            headers={\"Authorization\": f\"{token['token_type']} {token['access_token']}\"},\n        )\n        try:\n            response.raise_for_status()\n        except RequestException as exc:\n            LOGGER.warning(\n                \"Unable to fetch user profile\",\n                exc=exc,\n                response=exc.response.text if exc.response else str(exc),\n            )\n            return None\n        return response.json()\n", "path": "authentik/sources/oauth/clients/oauth2.py"}]}