Add support for roles in authorize calls

Author: RobertoPrevato

CHANGELOG.md

@@ -5,11 +5,14 @@ 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.3] - 2025-10-03
+
+- Add a `roles` property to the `Identity` object and a `RolesRequirement`
+  class to require roles.
+- Add support for validating JWTs signed using symmetric encryption
+  (`SymmetricJWTValidator` and `AsymmetricJWTValidator`).
+- Add support to call the `authorize` method with an optional set of roles,
+  treated as sufficient roles to succeed authorization.
 
 ## [1.0.2] - 2023-06-16 :corn:
 

guardpost/authorization.py

@@ -7,6 +7,7 @@
 
 from guardpost.abc import BaseStrategy
 from guardpost.authentication import Identity
+from guardpost.common import RolesRequirement
 
 
 class AuthorizationError(Exception):
@@ -208,46 +209,78 @@ def with_default_policy(self, policy: Policy) -> "AuthorizationStrategy":
         return self
 
     async def authorize(
-        self, policy_name: Optional[str], identity: Identity, scope: Any = None
+        self,
+        policy_name: Optional[str],
+        identity: Identity,
+        scope: Any = None,
+        roles: Optional[Sequence[str]] = None,
     ):
         if policy_name:
             policy = self.get_policy(policy_name)
 
             if not policy:
                 raise PolicyNotFoundError(policy_name)
 
-            await self._handle_with_policy(policy, identity, scope)
+            await self._handle_with_policy(policy, identity, scope, roles)
         else:
             if self.default_policy:
-                await self._handle_with_policy(self.default_policy, identity, scope)
+                await self._handle_with_policy(
+                    self.default_policy, identity, scope, roles
+                )
+                return
+
+            if roles:
+                # This code is only executed if the user specified roles without
+                # specifying an authorization policy.
+                await self._handle_with_roles(identity, roles)
                 return
 
             if not identity:
                 raise UnauthorizedError("Missing identity", [])
             if not identity.is_authenticated():
                 raise UnauthorizedError("The resource requires authentication", [])
 
-    def _get_requirements(self, policy: Policy, scope: Any) -> Iterable[Requirement]:
+    def _get_requirements(
+        self, policy: Policy, scope: Any, roles: Optional[Sequence[str]] = None
+    ) -> Iterable[Requirement]:
+        if roles:
+            yield RolesRequirement(roles=roles)
         yield from self._get_instances(policy.requirements, scope)
 
-    async def _handle_with_policy(self, policy: Policy, identity: Identity, scope: Any):
+    async def _handle_with_policy(
+        self,
+        policy: Policy,
+        identity: Identity,
+        scope: Any,
+        roles: Optional[Sequence[str]] = None,
+    ):
         with AuthorizationContext(
-            identity, list(self._get_requirements(policy, scope))
+            identity, list(self._get_requirements(policy, scope, roles))
         ) as context:
-            for requirement in context.requirements:
-                if _is_async_handler(type(requirement)):  # type: ignore
-                    await requirement.handle(context)
-                else:
-                    requirement.handle(context)  # type: ignore
-
-            if not context.has_succeeded:
-                if identity and identity.is_authenticated():
-                    raise ForbiddenError(
-                        context.forced_failure, context.pending_requirements
-                    )
-                raise UnauthorizedError(
+            await self._handle_context(identity, context)
+
+    async def _handle_with_roles(
+        self, identity: Identity, roles: Optional[Sequence[str]] = None
+    ):
+        # This method is to be used only when the user specified roles without a policy
+        with AuthorizationContext(identity, [RolesRequirement(roles=roles)]) as context:
+            await self._handle_context(identity, context)
+
+    async def _handle_context(self, identity: Identity, context: AuthorizationContext):
+        for requirement in context.requirements:
+            if _is_async_handler(type(requirement)):  # type: ignore
+                await requirement.handle(context)
+            else:
+                requirement.handle(context)  # type: ignore
+
+        if not context.has_succeeded:
+            if identity and identity.is_authenticated():
+                raise ForbiddenError(
                     context.forced_failure, context.pending_requirements
                 )
+            raise UnauthorizedError(
+                context.forced_failure, context.pending_requirements
+            )
 
     async def _handle_with_identity_getter(
         self, policy_name: Optional[str], *args, **kwargs

guardpost/common.py

@@ -69,17 +69,19 @@ def handle(self, context: AuthorizationContext):
 class RolesRequirement(Requirement):
     """
     Requires an identity with certain roles.
+    Supports defining sufficient roles (any one is enough), and required roles (all
+    must be present).
     """
 
-    __slots__ = ("_required_roles", "_sufficient_roles")
+    __slots__ = ("_roles", "_required_roles")
 
     def __init__(
         self,
+        roles: Optional[Sequence[str]] = None,
         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 [])
+        self._required_roles = list(required_roles) if required_roles else None
+        self._roles = list(roles) if roles else None
 
     def handle(self, context: AuthorizationContext):
         identity = context.identity
@@ -88,10 +90,10 @@ def handle(self, context: AuthorizationContext):
             context.fail("Missing identity")
             return
 
-        if self._required_roles:
-            if all(identity.has_role(name) for name in self._required_roles):
+        if self._roles:
+            if any(identity.has_role(name) for name in self._roles):
                 context.succeed(self)
 
-        if self._sufficient_roles:
-            if any(identity.has_role(name) for name in self._required_roles):
+        if self._required_roles:
+            if all(identity.has_role(name) for name in self._required_roles):
                 context.succeed(self)