new AuthClient (macrobond#82)

Author: mb-jp

.github/workflows/ci.yml

@@ -54,6 +54,10 @@ jobs:
       if: always()
       run: python scripts/test_setup.py
 
+    - name: Test pytest no_account
+      if: always()
+      run: python -m pytest -m no_account
+
     - name: Pdoc3
       if: always()
       run: python scripts/lint_tools.py --pdoc3

.vscode/launch.json

@@ -1,6 +1,18 @@
 {
     "version": "0.2.0",
     "configurations": [
+        {
+            "name": "Python: Debug Tests",
+            "type": "python",
+            "request": "launch",
+            "program": "${file}",
+            "purpose": ["debug-test"],
+            "console": "integratedTerminal",
+            "justMyCode": false,
+            "presentation": {
+                "hidden": true, // keep original launch order in 'run and debug' tab
+            }
+        }
     ],
     "compounds": [],
 }

README.md

@@ -90,16 +90,6 @@ python -m pip install macrobond-data-api
 
 Macrobond Data API for Python officially supports Python 3.8+.
 
-## Using of system keyring for http proxy
-
-For users operating behind an HTTP proxy, it is advisable to utilize the system keyring to store proxy settings and
-credentials.
-This can be conveniently accomplished by executing the included script with the following command:
-
-```console
-python -c "from macrobond_data_api.util import *; save_proxy_to_keyring()"
-```
-
 ## Using of system keyring for credentials
 
 > [!NOTE] 
@@ -112,6 +102,16 @@ This can be done easily by running the include script using this command:
 python -c "from macrobond_data_api.util import *; save_credentials_to_keyring()"
 ```
 
+## Using of system keyring for http proxy
+
+For users operating behind an HTTP proxy, it is advisable to utilize the system keyring to store proxy settings and
+credentials.
+This can be conveniently accomplished by executing the included script with the following command:
+
+```console
+python -c "from macrobond_data_api.util import *; save_proxy_to_keyring()"
+```
+
 ### Supported keyrings
 
 * macOS [Keychain](https://en.wikipedia.org/wiki/Keychain_%28software%29)

macrobond_data_api/_get_api.py

@@ -3,7 +3,7 @@
 from typing import TYPE_CHECKING, Optional, List
 
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: no cover
     from macrobond_data_api.common.api import Api
 
 __MACROBOND_DATA_API_CURRENT_API: Optional["Api"] = None

macrobond_data_api/web/__init__.py

@@ -3,3 +3,9 @@
 from .configuration import Configuration
 from .web_client import WebClient
 from .data_package_list_poller import DataPackageListPoller
+from .auth_exceptions import (
+    AuthBaseException,
+    AuthDiscoveryException,
+    AuthFetchTokenException,
+    AuthInvalidCredentialsException,
+)

macrobond_data_api/web/_auth_client.py

@@ -0,0 +1,116 @@
+import time
+from typing import Any, Dict, Sequence, Optional, TYPE_CHECKING
+
+from .auth_exceptions import AuthFetchTokenException, AuthDiscoveryException, AuthInvalidCredentialsException
+
+if TYPE_CHECKING:  # pragma: no cover
+    from .scope import Scope
+    from .session import Session
+    from requests.models import PreparedRequest
+
+
+class _AuthClient:
+
+    def __init__(
+        self, username: str, password: str, scope: Sequence["Scope"], authorization_url: str, session: "Session"
+    ) -> None:
+        self._username = username
+        self._password = password
+        self.scope = " ".join((str(x) for x in scope))
+        self.authorization_url = authorization_url
+        self.token_endpoint: Optional[str] = None
+        self.session = session
+        self.requests_auth = self._requests_auth
+        self.access_token = ""
+        self.token_response: Optional[Dict[str, Any]] = None
+        self.leeway = 60
+        self.expires_at: Optional[int] = None
+        self.fetch_token_get_time = time.time
+        self.is_expired_get_time = time.time
+
+    def fetch_token_if_necessary(self) -> bool:
+        if self._is_expired():
+            self.fetch_token()
+            return True
+        return False
+
+    def fetch_token(self) -> None:
+        if self.token_endpoint is None:
+            self.token_endpoint = self._discovery(self.authorization_url)
+
+        self._fetch_token(self.token_endpoint)
+
+    def _fetch_token(self, token_endpoint: str) -> None:
+        payload = {
+            "grant_type": "client_credentials",
+            "client_id": self._username,
+            "client_secret": self._password,
+            "scope": self.scope,
+        }
+        response = self.session.requests_session.post(
+            token_endpoint, data=payload, headers={"Content-Type": "application/x-www-form-urlencoded"}
+        )
+
+        if response.status_code not in [200, 400]:
+            raise AuthFetchTokenException("status code is not 200 or 400")
+
+        try:
+            json = self.token_response = response.json()
+        except Exception as ex:
+            raise AuthFetchTokenException("not valid json") from ex
+
+        if not isinstance(json, dict):
+            raise AuthFetchTokenException("no root obj in json")
+
+        if response.status_code == 400:
+            error = json.get("error")
+            if error == "invalid_client":
+                raise AuthInvalidCredentialsException("invalid client credentials")
+            if isinstance(error, str):
+                raise AuthFetchTokenException("error: " + error)
+            raise AuthFetchTokenException("no error in response")
+
+        if json.get("token_type") and json["token_type"] != "Bearer":
+            raise AuthFetchTokenException("token_type is not Bearer")
+
+        if json.get("expires_at"):
+            self.expires_at = int(json["expires_at"])
+        elif json.get("expires_in"):
+            self.expires_at = int(self.fetch_token_get_time()) + int(json["expires_in"])
+        else:
+            raise AuthFetchTokenException("no expires_at or expires_in")
+
+        if not json.get("access_token"):
+            raise AuthFetchTokenException("No access_token")
+
+        self.access_token = json["access_token"]
+
+    def _discovery(self, url: str) -> str:
+
+        response = self.session.requests_session.get(url + ".well-known/openid-configuration")
+        if response.status_code != 200:
+            raise AuthDiscoveryException("status code is not 200")
+
+        try:
+            json: Optional[Dict[str, Any]] = response.json()
+        except Exception as ex:
+            raise AuthDiscoveryException("not valid json.") from ex
+
+        if not isinstance(json, dict):
+            raise AuthDiscoveryException("no root obj in json.")
+
+        token_endpoint: Optional[str] = json.get("token_endpoint")
+        if token_endpoint is None:
+            raise AuthDiscoveryException("token_endpoint in root obj.")
+
+        return token_endpoint
+
+    def _is_expired(self) -> bool:
+        if not self.expires_at:
+            return True
+        expiration_threshold = self.expires_at - self.leeway
+        return expiration_threshold < self.is_expired_get_time()
+
+    def _requests_auth(self, r: "PreparedRequest") -> "PreparedRequest":
+        r.headers["Authorization"] = "Bearer " + self.access_token
+        return r

macrobond_data_api/web/auth_exceptions.py

@@ -0,0 +1,14 @@
+class AuthBaseException(Exception):
+    pass
+
+
+class AuthDiscoveryException(AuthBaseException):
+    pass
+
+
+class AuthFetchTokenException(AuthBaseException):
+    pass
+
+
+class AuthInvalidCredentialsException(AuthFetchTokenException):
+    pass