Fixing Not checking for expired JWT token

kiota_abstractions/authentication/base_bearer_token_authentication_provider.py

@@ -4,7 +4,11 @@
 # See License in the project root for license information.
 # ------------------------------------------------------------------------------
 
-from typing import Any, Dict
+import re
+import time
+from typing import Any, Dict, Optional
+
+import jwt
 
 from ..headers_collection import HeadersCollection
 from ..request_information import RequestInformation
@@ -13,18 +17,19 @@
 
 
 class BaseBearerTokenAuthenticationProvider(AuthenticationProvider):
-    """Provides a base class for implementing AuthenticationProvider for Bearer token scheme.
-    """
+    """Provides a base class for implementing AuthenticationProvider for Bearer token scheme."""
+
     AUTHORIZATION_HEADER = "Authorization"
     CLAIMS_KEY = "claims"
+    AUTHORIZATION_PREFIX = "Bearer"
 
     def __init__(self, access_token_provider: AccessTokenProvider) -> None:
         self.access_token_provider = access_token_provider
 
     async def authenticate_request(
         self,
         request: RequestInformation,
-        additional_authentication_context: Dict[str, Any] = {}
+        additional_authentication_context: Dict[str, Any] = None,
     ) -> None:
         """Authenticates the provided RequestInformation instance using the provided
         authorization token
@@ -34,21 +39,90 @@ async def authenticate_request(
         """
         if not request:
             raise Exception("Request cannot be null")
-        if all(
-            [
-                additional_authentication_context, self.CLAIMS_KEY
-                in additional_authentication_context,
-                request.headers.contains(self.AUTHORIZATION_HEADER)
-            ]
-        ):
-            request.headers.remove(self.AUTHORIZATION_HEADER)
 
         if not request.request_headers:
             request.headers = HeadersCollection()
 
+        if additional_authentication_context:
+            self._check_for_claims_key(request, additional_authentication_context)
+
+        self._remove_expired_token(request)
+
         if not request.headers.contains(self.AUTHORIZATION_HEADER):
             token = await self.access_token_provider.get_authorization_token(
                 request.url, additional_authentication_context
             )
             if token:
-                request.headers.add(f'{self.AUTHORIZATION_HEADER}', f'Bearer {token}')
+                request.headers.add(
+                    f"{self.AUTHORIZATION_HEADER}",
+                    f"{self.AUTHORIZATION_PREFIX} {token}",
+                )
+            return
+
+    def _check_for_claims_key(
+        self,
+        request: RequestInformation,
+        additional_authentication_context: Dict[str, Any],
+    ) -> None:
+        """
+        Checks if the claims key is in the additional authentication context and if the 
+        authorization header is in the request headers. If both conditions are met, it removes 
+        the authorization header from the request headers.
+
+        Args:
+            request (RequestInformation): The request information object.
+            additional_authentication_context (Dict[str, Any]): Additional context for authentication.
+        """
+
+        if all(
+            [
+                self.CLAIMS_KEY in additional_authentication_context,
+                request.headers.contains(self.AUTHORIZATION_HEADER),
+            ]
+        ):
+            request.headers.remove(self.AUTHORIZATION_HEADER)
+
+    def _remove_expired_token(self, request: RequestInformation) -> None:
+        """
+        Removes expired tokens from the request headers.
+
+        Args:
+            request (RequestInformation): The request information object.
+        """
+
+        tokens = request.headers.get(self.AUTHORIZATION_HEADER)
+
+        if tokens:
+            for _token in tokens:
+                matchs = re.match(rf"{self.AUTHORIZATION_PREFIX} (.*)", _token)
+                if matchs:
+                    token = matchs.group(0).split(" ")[1]
+                    if self.is_token_expired(token):
+                        request.headers.remove(self.AUTHORIZATION_HEADER)
+
+    @staticmethod
+    def is_token_expired(token: str) -> bool:
+        """
+        Checks if the given token is expired.
+
+        Args:
+            token (str): The token to check.
+
+        Returns:
+            bool: True if the token is expired, False otherwise.
+        """
+
+        try:
+            payload: Dict[str, Any] = jwt.decode(
+                token, options={"verify_signature": False}
+            )
+
+            # Get the expiration time (exp), which is a Unix timestamp
+            exp: Optional[int] = payload.get("exp", None)
+
+            # Check if the current time is past the token's expiration
+            if exp is not None and time.time() >= exp:
+                return True
+            return False
+        except jwt.DecodeError:
+            return True  # If we can't decode the token, consider it as expired