Add support for JSON payload in OAuth2 Client Credentials flow

requests_auth/_oauth2/authorization_code.py

@@ -151,10 +151,10 @@ def request_new_token(self):
         # As described in https://tools.ietf.org/html/rfc6749#section-4.1.4
         token, expires_in, refresh_token = request_new_grant_with_post(
             self.token_url,
-            self.token_data,
             self.token_field_name,
             self.timeout,
             self.session,
+            data=self.token_data,
         )
         # Handle both Access and Bearer tokens
         return (
@@ -168,10 +168,10 @@ def refresh_token(self, refresh_token: str):
         self.refresh_data["refresh_token"] = refresh_token
         token, expires_in, refresh_token = request_new_grant_with_post(
             self.token_url,
-            self.refresh_data,
             self.token_field_name,
             self.timeout,
             self.session,
+            data=self.refresh_data,
         )
         return self.state, token, expires_in, refresh_token
 

requests_auth/_oauth2/authorization_code_pkce.py

@@ -162,10 +162,10 @@ def request_new_token(self) -> tuple:
         # As described in https://tools.ietf.org/html/rfc6749#section-4.1.4
         token, expires_in, refresh_token = request_new_grant_with_post(
             self.token_url,
-            self.token_data,
             self.token_field_name,
             self.timeout,
             self.session,
+            data=self.token_data,
         )
         # Handle both Access and Bearer tokens
         return (
@@ -179,10 +179,10 @@ def refresh_token(self, refresh_token: str):
         self.refresh_data["refresh_token"] = refresh_token
         token, expires_in, refresh_token = request_new_grant_with_post(
             self.token_url,
-            self.refresh_data,
             self.token_field_name,
             self.timeout,
             self.session,
+            data=self.refresh_data,
         )
         return self.state, token, expires_in, refresh_token
 

requests_auth/_oauth2/client_credentials.py

@@ -63,6 +63,12 @@ def __init__(self, token_url: str, client_id: str, client_secret: str, **kwargs)
 
         # As described in https://tools.ietf.org/html/rfc6749#section-4.4.2
         self.data = {"grant_type": "client_credentials"}
+        self.post_as_json = kwargs.pop("post_as_json", False)
+
+        if self.post_as_json:
+            self.data["client_id"] = self.client_id
+            self.data["client_secret"] = self.client_secret
+
         scope = kwargs.pop("scope", None)
         if scope:
             self.data["scope"] = " ".join(scope) if isinstance(scope, list) else scope
@@ -81,15 +87,26 @@ def __call__(self, r):
         r.headers[self.header_name] = self.header_value.format(token=token)
         return r
 
-    def request_new_token(self) -> tuple:
-        # As described in https://tools.ietf.org/html/rfc6749#section-4.4.3
-        token, expires_in, _ = request_new_grant_with_post(
+    def request_new_grant_with_post(self) -> (str, int, str):
+        if self.post_as_json:
+            return request_new_grant_with_post(
+                self.token_url,
+                self.token_field_name,
+                self.timeout,
+                self.session,
+                json=self.data,
+            )
+        return request_new_grant_with_post(
             self.token_url,
-            self.data,
             self.token_field_name,
             self.timeout,
             self.session,
+            data=self.data,
         )
+
+    def request_new_token(self) -> tuple:
+        # As described in https://tools.ietf.org/html/rfc6749#section-4.4.3
+        token, expires_in, _ = self.request_new_grant_with_post()
         # Handle both Access and Bearer tokens
         return (self.state, token, expires_in) if expires_in else (self.state, token)
 

requests_auth/_oauth2/common.py

@@ -46,10 +46,13 @@ def _content_from_response(response: requests.Response) -> dict:
 
 
 def request_new_grant_with_post(
-    url: str, data, grant_name: str, timeout: float, session: requests.Session
+    url: str, grant_name: str, timeout: float, session: requests.Session, data: dict = None, json: dict = None
 ) -> (str, int, str):
     with session:
-        response = session.post(url, data=data, timeout=timeout)
+        if data is not None:
+            response = session.post(url, data=data, timeout=timeout)
+        else:
+            response = session.post(url, json=json, timeout=timeout)
         if not response:
             # As described in https://tools.ietf.org/html/rfc6749#section-5.2
             raise InvalidGrantRequest(response)

requests_auth/_oauth2/resource_owner_password.py

@@ -98,10 +98,10 @@ def request_new_token(self):
         # As described in https://tools.ietf.org/html/rfc6749#section-4.3.3
         token, expires_in, refresh_token = request_new_grant_with_post(
             self.token_url,
-            self.data,
             self.token_field_name,
             self.timeout,
             self.session,
+            data=self.data,
         )
         # Handle both Access and Bearer tokens
         return (
@@ -115,10 +115,10 @@ def refresh_token(self, refresh_token: str):
         self.refresh_data["refresh_token"] = refresh_token
         token, expires_in, refresh_token = request_new_grant_with_post(
             self.token_url,
-            self.refresh_data,
             self.token_field_name,
             self.timeout,
             self.session,
+            data=self.refresh_data,
         )
         return self.state, token, expires_in, refresh_token