Start improvements

Author: RobertoPrevato

CHANGELOG.md

@@ -5,20 +5,29 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.0.3] - 2025-10-02
+
+- Add a `roles` property to the `Identity` object and built-in classes for
+  role based authorization.
+- Add support for validating JWTs signed using symmetric encryption.
+
 ## [1.0.2] - 2023-06-16 :corn:
+
 - Raises a more specific exception `ForbiddenError` when the user of an
   operation is authenticated properly, but authorization fails.
   This enables better handling of authorization error, differentiating when the
   user context is missing or invalid, and when the context is valid but the
   user has no rights to do a certain operation. See [#371](https://github.com/Neoteroi/BlackSheep/issues/371).
 
 ## [1.0.1] - 2023-03-20 :sun_with_face:
+
 - Improves the automatic rotation of `JWKS`: when validating `JWTs`, `JWKS` are
   refreshed automatically if an unknown `kid` is encountered, and `JWKS` were
   last fetched more than `refresh_time` seconds ago (by default 120 seconds).
 - Corrects an inconsistency in how `claims` are read in the `User` class.
 
 ## [1.0.0] - 2023-01-07 :star:
+
 - Adds built-in support for dependency injection, using the new `ContainerProtocol`
   in `rodi` v2.
 - Removes the synchronous code API, maintaining only the asynchronous code API
@@ -29,24 +38,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Corrects `Identity.__getitem__` to raise `KeyError` if a claim is missing.
 
 ## [0.1.0] - 2022-11-06 :snake:
+
 - Workflow maintenance.
 
 ## [0.0.9] - 2021-11-14 :swan:
+
 - Adds `sub`, `access_token`, and `refresh_token` properties to the `Identity`.
   class
 - Adds `py.typed` file.
 
 ## [0.0.8] - 2021-10-31 :shield:
+
 - Adds classes to handle `JWT`s validation, but only for `RSA` keys.
 - Fixes issue (wrong arrangement in test) #5.
 - Includes `Python 3.10` in the CI/CD matrix.
 - Enforces `black` and `isort` in the CI pipeline.
 
 ## [0.0.7] - 2021-01-31 :grapes:
+
 - Corrects a bug in the `Policy` class (#2).
 - Changes the type annotation of `Identity` claims (#3).
 
 ## [0.0.6] - 2020-12-12 :octocat:
+
 - Completely migrates to GitHub Workflows.
 - Improves build to test Python 3.6 and 3.9.
 - Adds a changelog.

guardpost/authentication.py

@@ -28,6 +28,10 @@ def __init__(
     def sub(self) -> Optional[str]:
         return self.get("sub")
 
+    @property
+    def roles(self) -> Optional[str]:
+        return self.get("roles")
+
     def is_authenticated(self) -> bool:
         return bool(self.authentication_mode)
 
@@ -43,6 +47,11 @@ def has_claim(self, name: str) -> bool:
     def has_claim_value(self, name: str, value: str) -> bool:
         return self.claims.get(name) == value
 
+    def has_role(self, name: str) -> bool:
+        if not self.roles:
+            return False
+        return name in self.roles
+
 
 class User(Identity):
     @property

guardpost/common.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping
-from typing import Mapping as MappingType
+from typing import Mapping as MappingType, Optional
 from typing import Sequence, Union
 
 from .authorization import AuthorizationContext, Policy, Requirement
@@ -64,3 +64,34 @@ def handle(self, context: AuthorizationContext):
         else:
             if all(identity.has_claim(name) for name in self.required_claims):
                 context.succeed(self)
+
+
+class RolesRequirement(Requirement):
+    """
+    Requires an identity with certain roles.
+    """
+
+    __slots__ = ("_required_roles", "_sufficient_roles")
+
+    def __init__(
+        self,
+        required_roles: Optional[Sequence[str]] = None,
+        sufficient_roles: Optional[Sequence[str]] = None,
+    ):
+        self._required_roles = list(required_roles or [])
+        self._sufficient_roles = list(sufficient_roles or [])
+
+    def handle(self, context: AuthorizationContext):
+        identity = context.identity
+
+        if not identity:
+            context.fail("Missing identity")
+            return
+
+        if self._required_roles:
+            if all(identity.has_role(name) for name in self._required_roles):
+                context.succeed(self)
+
+        if self._sufficient_roles:
+            if any(identity.has_role(name) for name in self._required_roles):
+                context.succeed(self)

guardpost/jwts/__init__.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Optional, Sequence
+from typing import Any, Dict, Optional, Sequence, List, Union, Protocol
 
 import jwt
 from jwt.exceptions import InvalidIssuerError, InvalidTokenError
@@ -33,7 +33,29 @@ def get_kid(token: str) -> Optional[str]:
     return headers.get("kid")
 
 
-class JWTValidator:
+class JWTValidatorProtocol(Protocol):
+    """Protocol defining the interface for JWT validators"""
+
+    async def validate_jwt(self, access_token: str) -> Dict[str, Any]: ...
+
+
+class AbstractJWTValidator:
+    """Base class for JWT validators with common functionality"""
+
+    def __init__(
+        self,
+        *,
+        valid_issuers: Sequence[str],
+        valid_audiences: Sequence[str],
+        algorithms: Sequence[str],
+    ) -> None:
+        self._valid_issuers = list(valid_issuers)
+        self._valid_audiences = list(valid_audiences)
+        self._algorithms = list(algorithms)
+        self.logger = get_logger()
+
+
+class AsymmetricJWTValidator(AbstractJWTValidator):
     def __init__(
         self,
         *,
@@ -48,7 +70,7 @@ def __init__(
         refresh_time: float = 120,
     ) -> None:
         """
-        Creates a new instance of JWTValidator. This class only supports validating
+        Creates a new instance of AsymmetricJWTValidator. This class supports validating
         access tokens signed using asymmetric keys and handling JWKs of RSA type.
 
         Parameters
@@ -83,6 +105,12 @@ def __init__(
             JWKS were last fetched more than `refresh_time` seconds ago (by default
             120 seconds)
         """
+        super().__init__(
+            valid_issuers=valid_issuers,
+            valid_audiences=valid_audiences,
+            algorithms=algorithms,
+        )
+
         if keys_provider:
             pass
         elif authority:
@@ -96,14 +124,10 @@ def __init__(
                 "`authority`, or `keys_provider`."
             )
 
-        keys_provider = CachingKeysProvider(keys_provider, cache_time, refresh_time)
-
-        self._valid_issuers = list(valid_issuers)
-        self._valid_audiences = list(valid_audiences)
-        self._algorithms = list(algorithms)
-        self._keys_provider = keys_provider
+        self._keys_provider = CachingKeysProvider(
+            keys_provider, cache_time, refresh_time
+        )
         self.require_kid = require_kid
-        self.logger = get_logger()
 
     async def get_jwks(self) -> JWKS:
         return await self._keys_provider.get_keys()
@@ -170,3 +194,110 @@ async def validate_jwt(self, access_token: str) -> Dict[str, Any]:
                 return data
 
         raise InvalidAccessToken()
+
+
+class SymmetricJWTValidator(AbstractJWTValidator):
+    def __init__(
+        self,
+        *,
+        valid_issuers: Sequence[str],
+        valid_audiences: Sequence[str],
+        secret_key: Union[str, bytes],
+        algorithms: Sequence[str] = ["HS256"],
+    ) -> None:
+        """
+        Creates a new instance of SymmetricJWTValidator. This class supports validating
+        access tokens signed using symmetric keys (HMAC).
+
+        Parameters
+        ----------
+        valid_issuers : Sequence[str]
+            Sequence of acceptable issuers (iss).
+        valid_audiences : Sequence[str]
+            Sequence of acceptable audiences (aud).
+        secret_key : Union[str, bytes]
+            The secret key used for symmetric validation.
+        algorithms : Sequence[str], optional
+            Sequence of acceptable algorithms, by default ["HS256"].
+            Supported algorithms: HS256, HS384, HS512
+        """
+        super().__init__(
+            valid_issuers=valid_issuers,
+            valid_audiences=valid_audiences,
+            algorithms=algorithms,
+        )
+
+        supported_algorithms = ["HS256", "HS384", "HS512"]
+        for algorithm in algorithms:
+            if algorithm not in supported_algorithms:
+                raise ValueError(
+                    f"Algorithm '{algorithm}' is not supported for symmetric validation. "
+                    f"Use one of: {', '.join(supported_algorithms)}"
+                )
+
+        self._secret_key = secret_key
+
+    async def validate_jwt(self, access_token: str) -> Dict[str, Any]:
+        """
+        Validates the given JWT using symmetric key and returns its payload.
+        This method throws exception if the JWT is not valid.
+        """
+        for issuer in self._valid_issuers:
+            try:
+                return jwt.decode(
+                    access_token,
+                    self._secret_key,
+                    verify=True,
+                    algorithms=self._algorithms,
+                    audience=self._valid_audiences,
+                    issuer=issuer,
+                )
+            except InvalidIssuerError:
+                # Try the next issuer
+                pass
+            except InvalidTokenError as exc:
+                self.logger.debug("Invalid access token: ", exc_info=exc)
+
+        # If we've tried all issuers and none worked
+        raise InvalidAccessToken()
+
+
+class CompositeJWTValidator(AbstractJWTValidator):
+    def __init__(self, validators: List[JWTValidatorProtocol]) -> None:
+        """
+        Creates a composite validator that tries multiple validation strategies.
+        Useful when you need to support both symmetric and asymmetric validation.
+
+        Parameters
+        ----------
+        validators : List[JWTValidatorProtocol]
+            List of validators to try in sequence
+        """
+        self._validators = validators
+        self.logger = get_logger()
+
+    async def validate_jwt(self, access_token: str) -> Dict[str, Any]:
+        """
+        Attempts to validate the JWT using each validator in sequence.
+        Returns the first successful validation result or raises InvalidAccessToken
+        if all validators fail.
+        """
+        exceptions = []
+
+        for validator in self._validators:
+            try:
+                return await validator.validate_jwt(access_token)
+            except InvalidAccessToken as exc:
+                exceptions.append(exc)
+                # Continue to the next validator
+
+        # If we get here, all validators failed
+        if exceptions:
+            self.logger.debug(f"All validators failed: {exceptions}")
+        raise InvalidAccessToken(
+            "Token validation failed with all configured validators"
+        )
+
+
+# For backward compatibility, keep the original name
+JWTValidator = AsymmetricJWTValidator