[FEAT] SDK login directly to Auth0 [#483] (#69)

* Replaced login via pasqalcloud to auth0
* add more documentation
* see changelog
---------

Co-authored-by: Kilian Beuchard <[email protected]>

CHANGELOG.md

@@ -2,6 +2,19 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.1.15] 2023-03-27
+
+### Added
+    - Added tests to check the login behavior.
+    - Added tests to check the override Endpoints behavior.
+
+### Changed
+    - The authentication now directly connects to the Auth0 platform instead of connecting through PasqalCloud.
+    - Small refactor of files, with the authentication modules in the `authentication.py` file, instead of `client.py`.
+
+### Deleted
+    - Account endpoint, we now use Auth0.
+
 ## [0.1.14]
 
 ### Changed

README.md

@@ -29,6 +29,67 @@ To run the tutorials or the test suite locally, run the following to install the
 pip install -e .[dev]
 ```
 
+## Instanciation and Authentication
+
+There are several ways to provide a correct authentication using the SDK.
+
+```python
+from sdk import SDK
+
+group_id="your_group_id" # Replace this value by your group_id on the PASQAL platform.
+username="your_username" # Replace this value by your username or email on the PASQAL platform.
+password="your_password" # Replace this value by your password on the PASQAL platform.
+# Ideally, do not write this password in a script but provide in through the command-line or as a secret environment variable.
+
+""" Method 1: Username + Password
+    If you know your credentials, you can pass them to the SDK instance on creation.
+"""
+sdk = SDK(username=username, password=password, group_id=group_id)
+
+""" Method 2: Username only
+    If you only want to insert your username, but want a solution to have your password being secret
+    you can run the SDK without password. A prompt will then ask for your password
+"""
+sdk = SDK(username=username, group_id=group_id)
+> Please, enter your password:
+
+""" Method 3: Use a token
+    If you already know your token, you can directly pass it as an argument
+    using the following method.
+"""
+class NewTokenProvider(TokenProvider):
+    def _query_token(self):
+        # Custom token query that will be validated by the API Calls later.
+        return {
+            "access_token": "some_token",
+            "id_token": "id_token",
+            "scope": "openid profile email",
+            "expires_in": 86400,
+            "token_type": "Bearer"
+        }
+
+sdk = SDK(token_provider=NewTokenProvider, group_id=group_id)
+```
+
+/!\ For developers /!\
+
+If you want to redefine the APIs used by the SDK, please, do the following.
+
+```python
+from sdk import SDK, Endpoints, Auth0COnf
+
+endpoints = Endpoints(core = "my_new_core_endpoint")
+auth0 = Auth0Conf(
+    domain="new_domain",
+    public_client_id="public_id",
+    audience="new_audience"
+)
+sdk = SDK(..., endpoints=endpoints, auth0=auth0)
+```
+
+This enables you to target backend services running locally on your machine, or other environment like preprod or dev.
+
+
 ## Basic usage
 
 The package main component is a python object called `SDK` which can be used to create a `Batch` and send it automatically
@@ -66,11 +127,9 @@ Once you have serialized your sequence, you can send it with the SDK with the fo
 from sdk import SDK
 from pulser import devices, Register, Sequence
 
-username="your_username" # Replace this value by your username or email on the PASQAL platform.
 group_id="your_group_id" # Replace this value by your group_id on the PASQAL platform.
+username="your_username" # Replace this value by your username or email on the PASQAL platform.
 password="your_password" # Replace this value by your password on the PASQAL platform.
-# Ideally, do not write this password in a script but provide in through the command-line or as a secret environment variable.
-
 
 sdk = SDK(username=username, password=password, group_id=group_id)
 

mypy.ini

@@ -8,4 +8,4 @@ warn_unused_ignores = true
 ignore_errors = false
 warn_redundant_casts = true
 allow_untyped_calls = true
-allow_any_generics = true
+allow_any_generics = true

sdk/__init__.py

@@ -15,11 +15,12 @@
 import time
 from typing import Any, Dict, List, Optional
 
+from sdk.authentication import TokenProvider
 from sdk.batch import Batch, RESULT_POLLING_INTERVAL
-from sdk.client import Client, TokenProvider
+from sdk.client import Client
+from sdk.endpoints import Endpoints, Auth0Conf
 from sdk.device.configuration import BaseConfig
 from sdk.device.device_types import DeviceType
-from sdk.endpoints import Endpoints
 from sdk.job import Job
 
 
@@ -33,14 +34,22 @@ def __init__(
         password: Optional[str] = None,
         token_provider: Optional[TokenProvider] = None,
         endpoints: Optional[Endpoints] = None,
+        auth0: Optional[Auth0Conf] = None,
         webhook: Optional[str] = None,
     ):
+        """This class provides helper methods to call the PASQAL Cloud endpoints.
+
+        To authenticate to PASQAL Cloud, you have to provide either an
+        email/password combination or a TokenProvider instance.
+        You may omit the password, you will then be prompted to enter one.
+        """
         self._client = Client(
+            group_id=group_id,
             username=username,
             password=password,
-            group_id=group_id,
             token_provider=token_provider,
             endpoints=endpoints,
+            auth0=auth0,
         )
         self.batches: Dict[str, Batch] = {}
         self.webhook = webhook

sdk/_version.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__ = "0.1.14"
+__version__ = "0.1.15"

sdk/authentication.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from datetime import datetime, timedelta, timezone
+from jwt import decode, DecodeError
+from requests import PreparedRequest
+from typing import Any, Optional
+
+from auth0.v3.authentication import GetToken  # type: ignore
+from auth0.v3.exceptions import Auth0Error  # type: ignore
+from requests.auth import AuthBase
+
+from sdk.endpoints import Auth0Conf
+from sdk.errors import HTTPError
+
+
+class HTTPBearerAuthenticator(AuthBase):
+    def __init__(self, token_provider: Optional[TokenProvider]):
+        if not token_provider:
+            raise Exception("The authenticator needs a token provider.")
+        self.token_provider = token_provider
+
+    def __call__(self, r: PreparedRequest) -> PreparedRequest:
+        r.headers["Authorization"] = f"Bearer {self.token_provider.get_token()}"
+        return r
+
+
+class TokenProviderError(Exception):
+    pass
+
+
+class TokenProvider(ABC):
+    __token_cache: Optional[tuple[datetime, str]] = None
+    expiry_window: timedelta = timedelta(minutes=1.0)
+
+    def __init__(self, username: str, password: str, auth0: Auth0Conf):
+        self.username = username
+        self.password = password
+        self.auth0 = auth0
+
+    @abstractmethod
+    def _query_token(self) -> dict[str, Any]:
+        raise NotImplementedError
+
+    def get_token(self) -> str:
+        if self.__token_cache:
+            expiry, token = self.__token_cache
+            if expiry - self.expiry_window > datetime.now(tz=timezone.utc):
+                return token
+
+        self.__token_cache = self._refresh_token_cache()
+        return self.__token_cache[1]
+
+    def _refresh_token_cache(self) -> tuple[datetime, str]:
+        try:
+            token_response = self._query_token()
+            expiry = self._extract_expiry(token_response)
+
+            if expiry is None:
+                expiry = datetime.now(tz=timezone.utc)
+
+            return (expiry, token_response["access_token"])
+        except HTTPError as err:
+            raise TokenProviderError(err)
+
+    @staticmethod
+    def _extract_expiry(token_response: dict[str, Any]) -> datetime | None:
+        """Extracts the expiry datetime from the token response.
+
+        Assumes that the actual token provided is correct, but might be in an
+        unexpected format
+
+        Args:
+            token_response: A valid token response dictionary.
+
+        Returns:
+            The time the token will expire, or None if it can't be calculated
+        """
+        expires_in = token_response.get("expires_in", None)
+        if expires_in:
+            return datetime.now(tz=timezone.utc) + timedelta(seconds=float(expires_in))
+
+        try:
+            # We assume the token is valid, but might not be in an expected format
+            token = token_response.get("access_token")
+
+            if isinstance(token, str) and "." in token:
+                token_timestamp = decode(
+                    token,
+                    algorithms=["HS256"],
+                    options={"verify_signature": False},
+                ).get("exp")
+                if token_timestamp:
+                    return datetime.fromtimestamp(token_timestamp, tz=timezone.utc)
+
+        except DecodeError:
+            # The token is not in a format that could be parsed as a JWT
+            pass
+        return None
+
+
+class Auth0TokenProvider(TokenProvider):
+    def __init__(self, username: str, password: str, auth0: Auth0Conf):
+        super().__init__(username, password, auth0)
+
+        # Makes a call in order to check the credentials at creation
+        self.get_token()
+
+    def _query_token(self) -> dict[str, Any]:
+        token = GetToken(self.auth0.domain)
+
+        # No client secret required for this Application since
+        # "Token Endpoint Authentication Method" set to None
+        validated_token: dict[str, Any] = token.login(
+            client_id=self.auth0.public_client_id,
+            client_secret="",
+            username=self.username,
+            password=self.password,
+            audience=self.auth0.audience,
+            scope="openid profile email",
+            realm=self.auth0.realm,
+            grant_type="http://auth0.com/oauth/grant-type/password-realm",
+        )
+        return validated_token