feat: check that token is valid upon instantiation of config object (#28)

* feat: check that token is valid upon instantiation of config object

* linter

* fix coverage

Author: hf-kklein

src/bssclient/client/bssclient.py

@@ -11,7 +11,7 @@
 from yarl import URL
 
 from bssclient.client.config import BasicAuthBssConfig, BssConfig, OAuthBssConfig
-from bssclient.client.oauth import _OAuthHttpClient
+from bssclient.client.oauth import _OAuthHttpClient, token_is_valid
 from bssclient.models.aufgabe import AufgabeStats
 from bssclient.models.ermittlungsauftrag import Ermittlungsauftrag, _ListOfErmittlungsauftraege
 
@@ -210,7 +210,7 @@ async def _get_session(self) -> ClientSession:
         async with self._session_lock:
             if self._bearer_token is None:
                 self._bearer_token = await self._get_oauth_token()
-            elif not self._token_is_valid(self._bearer_token):
+            elif not token_is_valid(self._bearer_token):
                 await self.close_session()
             if self._session is None or self._session.closed:
                 _logger.info("creating new session")

src/bssclient/client/config.py

@@ -5,6 +5,8 @@
 from pydantic import BaseModel, ConfigDict, HttpUrl, field_validator, model_validator
 from yarl import URL
 
+from .oauth import token_is_valid
+
 
 class BssConfig(BaseModel):
     """
@@ -102,3 +104,14 @@ def check_secret_or_token_is_present(cls, values):  # pylint:disable=no-self-arg
                 # pylint:disable=line-too-long
                 "You must provide either client id and secret or a bearer token, but not None of both"
             )
+
+    @field_validator("bearer_token")
+    def validate_bearer_token(cls, value):
+        """
+        check that the value is a string
+        """
+        if value is not None and len(value.strip()) > 0:
+            _token_is_valid = token_is_valid(value)
+            if not _token_is_valid:
+                raise ValueError("Invalid bearer token")
+        return value

src/bssclient/client/oauth.py

@@ -15,6 +15,27 @@
 _logger = logging.getLogger(__name__)
 
 
+def token_is_valid(token) -> bool:
+    """
+    returns true iff the token expiration date is far enough in the future. By "enough" I mean:
+    more than 1 minute (because the clients' request using the token shouldn't take longer than that)
+    """
+    try:
+        decoded_token = jwt.decode(token, algorithms=["HS256"], options={"verify_signature": False})
+        expiration_timestamp = decoded_token.get("exp")
+        expiration_datetime = datetime.fromtimestamp(expiration_timestamp)
+        _logger.debug("Token is valid until %s", expiration_datetime.isoformat())
+        current_datetime = datetime.utcnow()
+        token_is_valid_one_minute_into_the_future = expiration_datetime > current_datetime + timedelta(minutes=1)
+        return token_is_valid_one_minute_into_the_future
+    except jwt.ExpiredSignatureError:
+        _logger.info("The token is expired", exc_info=True)
+        return False
+    except jwt.InvalidTokenError:
+        _logger.info("The token is invalid", exc_info=True)
+        return False
+
+
 class _ValidateTokenMixin:  # pylint:disable=too-few-public-methods
     """
     Mixin for classes which need to validate tokens
@@ -23,26 +44,6 @@ class _ValidateTokenMixin:  # pylint:disable=too-few-public-methods
     def __init__(self):
         self._session_lock = asyncio.Lock()
 
-    def _token_is_valid(self, token) -> bool:
-        """
-        returns true iff the token expiration date is far enough in the future. By "enough" I mean:
-        more than 1 minute (because the clients' request using the token shouldn't take longer than that)
-        """
-        try:
-            decoded_token = jwt.decode(token, algorithms=["HS256"], options={"verify_signature": False})
-            expiration_timestamp = decoded_token.get("exp")
-            expiration_datetime = datetime.fromtimestamp(expiration_timestamp)
-            _logger.debug("Token is valid until %s", expiration_datetime.isoformat())
-            current_datetime = datetime.utcnow()
-            token_is_valid_one_minute_into_the_future = expiration_datetime > current_datetime + timedelta(minutes=1)
-            return token_is_valid_one_minute_into_the_future
-        except jwt.ExpiredSignatureError:
-            _logger.info("The token is expired", exc_info=True)
-            return False
-        except jwt.InvalidTokenError:
-            _logger.info("The token is invalid", exc_info=True)
-            return False
-
 
 class _OAuthHttpClient(_ValidateTokenMixin, ABC):  # pylint:disable=too-few-public-methods
     """
@@ -86,7 +87,7 @@ async def _get_oauth_token(self) -> str:
             if self._token is None:
                 _logger.info("Initially retrieving a new token")
                 self._token = await self._get_new_token()
-            elif not self._token_is_valid(self._token):
+            elif not token_is_valid(self._token):
                 _logger.info("Token is not valid anymore, retrieving a new token")
                 self._token = await self._get_new_token()
             else:

unittests/test_bss_client.py

@@ -2,7 +2,7 @@
 from yarl import URL
 
 from bssclient.client.bssclient import BasicAuthBssClient
-from bssclient.client.config import BasicAuthBssConfig
+from bssclient.client.config import BasicAuthBssConfig, OAuthBssConfig
 
 
 @pytest.mark.parametrize(
@@ -22,3 +22,8 @@ def test_get_tld(actual_url: URL, expected_tld: URL):
     client = BasicAuthBssClient(config)
     actual = client.get_top_level_domain()
     assert actual == expected_tld
+
+
+def test_oauth_config():
+    with pytest.raises(ValueError):
+        OAuthBssConfig(server_url=URL("https://bss.example.com"), bearer_token="something-which-is-definittly no token")