Add WakaTimeAuthorizationCode and Browser display settings (#84)

Author: Colin-b

CHANGELOG.md

@@ -10,20 +10,34 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Publicly expose `requests_auth.SupportMultiAuth`, allowing multiple authentication support for every `requests` authentication class that exists.
 - Publicly expose `requests_auth.TokenMemoryCache`, allowing to create custom Oauth2 token cache based on this default implementation.
 - Thanks to the new `redirect_uri_domain` parameter on Authorization code (with and without PKCE) and Implicit flows, you can now provide the [FQDN](https://en.wikipedia.org/wiki/Fully_qualified_domain_name) to use in the `redirect_uri` when `localhost` (the default) is not allowed.
+- `requests_auth.WakaTimeAuthorizationCode` handling access to the [WakaTime API](https://wakatime.com/developers).
 
 ### Changed
 - Except for `requests_auth.testing`, only direct access via `requests_auth.` was considered publicly exposed. This is now explicit, as inner packages are now using private prefix (`_`).
   If you were relying on some classes or functions that are now internal, feel free to open an issue.
 - `requests_auth.JsonTokenFileCache` and `requests_auth.TokenMemoryCache` `get_token` method does not handle kwargs anymore, the `on_missing_token` callable does not expect any arguments anymore.
 - `requests_auth.JsonTokenFileCache` does not expose `tokens_path` or `last_save_time` attributes anymore and is also allowing `pathlib.Path` instances as cache location.
 - `requests_auth.TokenMemoryCache` does not expose `forbid_concurrent_cache_access` or `forbid_concurrent_missing_token_function_call` attributes anymore.
+- Browser display settings have been moved to a shared setting, see documentation for more information on `requests_auth.OAuth2.display`.
+  As a result the following classes no longer expose `success_display_time` and `failure_display_time` parameters.
+  - `requests_auth.OAuth2AuthorizationCode`.
+  - `requests_auth.OktaAuthorizationCode`.
+  - `requests_auth.WakaTimeAuthorizationCode`.
+  - `requests_auth.OAuth2AuthorizationCodePKCE`.
+  - `requests_auth.OktaAuthorizationCodePKCE`.
+  - `requests_auth.OAuth2Implicit`.
+  - `requests_auth.AzureActiveDirectoryImplicit`.
+  - `requests_auth.AzureActiveDirectoryImplicitIdToken`.
+  - `requests_auth.OktaImplicit`.
+  - `requests_auth.OktaImplicitIdToken`.
 
 ### Fixed
 - Type information is now provided following [PEP 561](https://www.python.org/dev/peps/pep-0561/).
 - Remove deprecation warnings due to usage of `utcnow` and `utcfromtimestamp`.
 - `requests_auth.OktaClientCredentials` `scope` parameter is now mandatory and does not default to `openid` anymore.
 - `requests_auth.OktaClientCredentials` will now display a more user-friendly error message in case Okta instance is not provided.
 - Tokens cache `DEBUG` logs will not display tokens anymore.
+- Handle `text/html; charset=utf-8` content-type in token responses.
 
 ### Removed
 - Removing support for Python `3.7`.

README.md

@@ -18,11 +18,9 @@ keywords = [
     "authentication",
     "ntlm",
     "oauth2",
-    "azure-active-directory",
-    "azure-ad",
     "okta",
-    "apikey",
-    "multiple",
+    "aad",
+    "entra"
 ]
 classifiers=[
     "Development Status :: 5 - Production/Stable",

pyproject.toml

@@ -5,10 +5,12 @@
     NTLM,
     SupportMultiAuth,
 )
+from requests_auth._oauth2.browser import DisplaySettings
 from requests_auth._oauth2.common import OAuth2
 from requests_auth._oauth2.authorization_code import (
     OAuth2AuthorizationCode,
     OktaAuthorizationCode,
+    WakaTimeAuthorizationCode,
 )
 from requests_auth._oauth2.authorization_code_pkce import (
     OAuth2AuthorizationCodePKCE,
@@ -46,6 +48,7 @@
     "HeaderApiKey",
     "QueryApiKey",
     "OAuth2",
+    "DisplaySettings",
     "OAuth2AuthorizationCodePKCE",
     "OktaAuthorizationCodePKCE",
     "OAuth2Implicit",
@@ -55,6 +58,7 @@
     "AzureActiveDirectoryImplicitIdToken",
     "OAuth2AuthorizationCode",
     "OktaAuthorizationCode",
+    "WakaTimeAuthorizationCode",
     "OAuth2ClientCredentials",
     "OktaClientCredentials",
     "OAuth2ResourceOwnerPasswordCredentials",

requests_auth/__init__.py

@@ -5,6 +5,8 @@
 from urllib.parse import parse_qs, urlparse
 from socket import socket
 
+from requests_auth._oauth2.common import OAuth2
+
 from requests_auth._errors import *
 
 logger = logging.getLogger(__name__)
@@ -86,7 +88,7 @@ def send_html(self, html_content: str):
         logger.debug("HTML content sent to client.")
 
     def success_page(self, text: str):
-        return f"""<body onload="window.open('', '_self', ''); window.setTimeout(close, {self.server.grant_details.reception_success_display_time})" style="
+        return f"""<body onload="window.open('', '_self', ''); window.setTimeout(close, {OAuth2.display.success_display_time})" style="
         color: #4F8A10;
         background-color: #DFF2BF;
         font-size: xx-large;
@@ -97,7 +99,7 @@ def success_page(self, text: str):
         </body>"""
 
     def error_page(self, text: str):
-        return f"""<body onload="window.open('', '_self', ''); window.setTimeout(close, {self.server.grant_details.reception_failure_display_time})" style="
+        return f"""<body onload="window.open('', '_self', ''); window.setTimeout(close, {OAuth2.display.failure_display_time})" style="
         color: #D8000C;
         background-color: #FFBABA;
         font-size: xx-large;
@@ -137,15 +139,11 @@ def __init__(
         url: str,
         name: str,
         reception_timeout: float,
-        reception_success_display_time: int,
-        reception_failure_display_time: int,
         redirect_uri_port: int,
     ):
         self.url = url
         self.name = name
         self.reception_timeout = reception_timeout
-        self.reception_success_display_time = reception_success_display_time
-        self.reception_failure_display_time = reception_failure_display_time
         self.redirect_uri_port = redirect_uri_port
 
 

requests_auth/_oauth2/authentication_responses_server.py

@@ -1,4 +1,6 @@
 from hashlib import sha512
+from typing import Union, Iterable
+
 import requests
 import requests.auth
 
@@ -36,12 +38,6 @@ def __init__(self, authorization_url: str, token_url: str, **kwargs):
         Listen on port 5000 by default.
         :param timeout: Maximum amount of seconds to wait for a code or a token to be received once requested.
         Wait for 1 minute by default.
-        :param success_display_time: In case a code is successfully received,
-        this is the maximum amount of milliseconds the success page will be displayed in your browser.
-        Display the page for 1 millisecond by default.
-        :param failure_display_time: In case received code is not valid,
-        this is the maximum amount of milliseconds the failure page will be displayed in your browser.
-        Display the page for 5 seconds by default.
         :param header_name: Name of the header field used to send token.
         Token will be sent in Authorization header field by default.
         :param header_value: Format used to send the token value.
@@ -120,8 +116,6 @@ def __init__(self, authorization_url: str, token_url: str, **kwargs):
             code_grant_url,
             code_field_name,
             self.timeout,
-            self.success_display_time,
-            self.failure_display_time,
             self.redirect_uri_port,
         )
 
@@ -211,12 +205,6 @@ def __init__(self, instance: str, client_id: str, **kwargs):
         Listen on port 5000 by default.
         :param timeout: Maximum amount of seconds to wait for a token to be received once requested.
         Wait for 1 minute by default.
-        :param success_display_time: In case a token is successfully received,
-        this is the maximum amount of milliseconds the success page will be displayed in your browser.
-        Display the page for 1 millisecond by default.
-        :param failure_display_time: In case received token is not valid,
-        this is the maximum amount of milliseconds the failure page will be displayed in your browser.
-        Display the page for 5 seconds by default.
         :param header_name: Name of the header field used to send token.
         Token will be sent in Authorization header field by default.
         :param header_value: Format used to send the token value.
@@ -239,3 +227,58 @@ def __init__(self, instance: str, client_id: str, **kwargs):
             client_id=client_id,
             **kwargs,
         )
+
+
+class WakaTimeAuthorizationCode(OAuth2AuthorizationCode):
+    """
+    Describes a WakaTime (OAuth 2) "Access Token" authorization code flow requests authentication.
+    """
+
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        scope: Union[str, Iterable[str]],
+        **kwargs,
+    ):
+        """
+        :param client_id: WakaTime Application Identifier (formatted as a Universal Unique Identifier)
+        :param client_secret: WakaTime Application Secret (formatted as waka_sec_ followed by a Universal Unique Identifier)
+        :param scope: Scope parameter sent in query. Can also be a list of scopes.
+        :param response_type: Value of the response_type query parameter.
+        token by default.
+        :param token_field_name: Name of the expected field containing the token.
+        access_token by default.
+        :param early_expiry: Number of seconds before actual token expiry where token will be considered as expired.
+        Default to 30 seconds to ensure token will not expire between the time of retrieval and the time the request
+        reaches the actual server. Set it to 0 to deactivate this feature and use the same token until actual expiry.
+        :param nonce: Refer to http://openid.net/specs/openid-connect-core-1_0.html#IDToken for more details
+        (formatted as a Universal Unique Identifier - UUID). Use a newly generated UUID by default.
+        :param redirect_uri_domain: FQDN to use in the redirect_uri when localhost (default) is not allowed.
+        :param redirect_uri_endpoint: Custom endpoint that will be used as redirect_uri the following way:
+        http://localhost:<redirect_uri_port>/<redirect_uri_endpoint>. Default value is to redirect on / (root).
+        :param redirect_uri_port: The port on which the server listening for the OAuth 2 token will be started.
+        Listen on port 5000 by default.
+        :param timeout: Maximum amount of seconds to wait for a token to be received once requested.
+        Wait for 1 minute by default.
+        :param header_name: Name of the header field used to send token.
+        Token will be sent in Authorization header field by default.
+        :param header_value: Format used to send the token value.
+        "{token}" must be present as it will be replaced by the actual token.
+        Token will be sent as "Bearer {token}" by default.
+        :param session: requests.Session instance that will be used to request the token.
+        Use it to provide a custom proxying rule for instance.
+        :param kwargs: all additional authorization parameters that should be put as query parameter
+        in the authorization URL.
+        """
+        if not scope:
+            raise Exception("Scope is mandatory.")
+        OAuth2AuthorizationCode.__init__(
+            self,
+            "https://wakatime.com/oauth/authorize",
+            "https://wakatime.com/oauth/token",
+            client_id=client_id,
+            client_secret=client_secret,
+            scope=",".join(scope) if isinstance(scope, list) else scope,
+            **kwargs,
+        )

requests_auth/_oauth2/authorization_code.py

@@ -40,12 +40,6 @@ def __init__(self, authorization_url: str, token_url: str, **kwargs):
         Listen on port 5000 by default.
         :param timeout: Maximum amount of seconds to wait for a code or a token to be received once requested.
         Wait for 1 minute by default.
-        :param success_display_time: In case a code is successfully received,
-        this is the maximum amount of milliseconds the success page will be displayed in your browser.
-        Display the page for 1 millisecond by default.
-        :param failure_display_time: In case received code is not valid,
-        this is the maximum amount of milliseconds the failure page will be displayed in your browser.
-        Display the page for 5 seconds by default.
         :param header_name: Name of the header field used to send token.
         Token will be sent in Authorization header field by default.
         :param header_value: Format used to send the token value.
@@ -131,8 +125,6 @@ def __init__(self, authorization_url: str, token_url: str, **kwargs):
             code_grant_url,
             code_field_name,
             self.timeout,
-            self.success_display_time,
-            self.failure_display_time,
             self.redirect_uri_port,
         )
 
@@ -257,12 +249,6 @@ def __init__(self, instance: str, client_id: str, **kwargs):
         Listen on port 5000 by default.
         :param timeout: Maximum amount of seconds to wait for a token to be received once requested.
         Wait for 1 minute by default.
-        :param success_display_time: In case a token is successfully received,
-        this is the maximum amount of milliseconds the success page will be displayed in your browser.
-        Display the page for 1 millisecond by default.
-        :param failure_display_time: In case received token is not valid,
-        this is the maximum amount of milliseconds the failure page will be displayed in your browser.
-        Display the page for 5 seconds by default.
         :param header_name: Name of the header field used to send token.
         Token will be sent in Authorization header field by default.
         :param header_value: Format used to send the token value.

requests_auth/_oauth2/authorization_code_pkce.py

@@ -28,3 +28,25 @@ def __init__(self, kwargs):
         self.failure_display_time = int(
             kwargs.pop("failure_display_time", None) or 5000
         )
+
+
+class DisplaySettings:
+    def __init__(
+        self,
+        *,
+        success_display_time: int = 1,
+        failure_display_time: int = 5_000,
+    ):
+        """
+        :param success_display_time: In case a code/token is successfully received,
+        this is the maximum amount of milliseconds the success page will be displayed in your browser.
+        Display the page for 1 millisecond by default.
+        :param failure_display_time: In case received code/token is not valid,
+        this is the maximum amount of milliseconds the failure page will be displayed in your browser.
+        Display the page for 5 seconds by default.
+        """
+        # Time is expressed in milliseconds
+        self.success_display_time = success_display_time
+
+        # Time is expressed in milliseconds
+        self.failure_display_time = failure_display_time

requests_auth/_oauth2/browser.py

@@ -4,6 +4,7 @@
 import requests.auth
 
 from requests_auth._errors import InvalidGrantRequest, GrantNotProvided
+from requests_auth._oauth2.browser import DisplaySettings
 from requests_auth._oauth2.tokens import TokenMemoryCache
 
 
@@ -33,6 +34,17 @@ def _get_query_parameter(url: str, param_name: str) -> Optional[str]:
     return all_values[0] if all_values else None
 
 
+def _content_from_response(response: requests.Response) -> dict:
+    content_type = response.headers.get("content-type")
+    if content_type == "text/html; charset=utf-8":
+        return {
+            key_values[0]: key_values[1]
+            for key_value in response.text.split("&")
+            if (key_values := key_value.split("=")) and len(key_values) == 2
+        }
+    return response.json()
+
+
 def request_new_grant_with_post(
     url: str, data, grant_name: str, timeout: float, session: requests.Session
 ) -> (str, int, str):
@@ -42,7 +54,7 @@ def request_new_grant_with_post(
             # As described in https://tools.ietf.org/html/rfc6749#section-5.2
             raise InvalidGrantRequest(response)
 
-        content = response.json()
+        content = _content_from_response(response)
     token = content.get(grant_name)
     if not token:
         raise GrantNotProvided(grant_name, content)
@@ -51,3 +63,4 @@ def request_new_grant_with_post(
 
 class OAuth2:
     token_cache = TokenMemoryCache()
+    display = DisplaySettings()