Sanic support (#4)

* async support

* sanic support

.github/workflows/validate-pull-request.yml

@@ -5,7 +5,7 @@ on: pull_request
 jobs:
   test:
     strategy:
-      fail-fast: true
+      fail-fast: false
       matrix:
         python-version: [ "3.7", "3.8", "3.9", "3.10" ]
     runs-on: ubuntu-latest
@@ -29,7 +29,7 @@ jobs:
           key: venv-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('**/poetry.lock') }}-1
       - name: Install dependencies
         if: steps.load-cache.outputs.cache-hit != 'true'
-        run: poetry install --no-interaction --no-root --extras "falcon"
+        run: poetry install --no-interaction --no-root --extras "falcon sanic"
       - name: black
         run: poetry run black --check .
       - name: isort

.pre-commit-config.yaml

@@ -3,14 +3,14 @@ repos:
   rev: 22.3.0
   hooks:
   - id: black
-- repo: https://github.com/PyCQA/flake8
-  rev: 4.0.1
-  hooks:
-  - id: flake8
 - repo: https://github.com/PyCQA/isort
   rev: 5.10.1
   hooks:
   - id: isort
+- repo: https://github.com/PyCQA/flake8
+  rev: 4.0.1
+  hooks:
+  - id: flake8
 - repo: https://github.com/pre-commit/mirrors-mypy
   rev: 'v0.950'
   hooks:

denied/__init__.py

@@ -1 +1,8 @@
 __version__ = "0.1.0"
+
+from ._async.ability import Ability
+from ._async.policy import Policy, authorize
+from .action import Action
+from .permission import AutoPermission, Permission
+
+__all__ = ["Ability", "Action", "Policy", "authorize", "Permission", "AutoPermission"]

denied/_async/ability.py

@@ -0,0 +1,55 @@
+from typing import Any, Optional
+
+from denied.action import Action
+from denied.errors import UnauthorizedError, UndefinedPermission
+from denied.permission import Permission
+
+from .policy import Policy
+
+
+class Ability:
+    def __init__(
+        self, policy: Optional[Policy] = None, default_action: Action = Action.DENY
+    ):
+        self._policy = policy or Policy()
+        self._default_action = default_action
+
+    async def authorize(
+        self, permission: Permission, *args: Any, **kwargs: Any
+    ) -> None:
+        """Raises an UnauthorizedError if policy does not grant permission.
+
+        Args:
+            permission (Permission): a permission
+            args (Any): arguments passed to the policy access method
+            kwargs (Any): keyword argumentss passed to the policy access method
+
+        Returns:
+            None:
+        """
+        if not await self.can(permission, *args, **kwargs):
+            raise UnauthorizedError(permission)
+
+    async def can(self, permission: Permission, *args: Any, **kwargs: Any) -> bool:
+        """Returns the result of the policy access method defined for the permission.
+        If no access method is found the default_action is used.
+        If permission was not defined and default_action is RAISE then an
+        UndefinedPermission is raised.
+
+        Args:
+            permission (Permission): a permission
+            args (Any): arguments passed to the policy access method
+            kwargs (Any): keyword argumentss passed to the policy access method
+
+        Returns:
+            bool: True if permission is granted, False otherwise
+        """
+        try:
+            access_method = self._policy.get_access_method(permission)
+        except UndefinedPermission as error:
+            if self._default_action == Action.RAISE:
+                raise error
+            else:
+                return True if self._default_action == Action.ALLOW else False
+
+        return await access_method(*args, **kwargs)

denied/policy.py → denied/_async/policy.py

@@ -1,24 +1,8 @@
 from typing import Any, Callable, Dict, Optional, Tuple
 
-from .errors import UndefinedPermission
-from .permission import Permission
-
-
-class PermissionAlreadyDefined(Exception):
-    """Error raised while using the @authorize() decorator with the same Permission
-    twice in the same Policy object.
-    """
-
-    def __init__(self, permission: Permission) -> None:
-        """
-        Args:
-            permission (Permission): a permission
-        """
-        super().__init__(f"Permission {permission.name} already defined")
-        self.permission = permission
-
-
-AccessMethod = Callable[..., bool]
+from denied.errors import PermissionAlreadyDefined, UndefinedPermission
+from denied.permission import Permission
+from denied.utils import AccessMethod
 
 
 class PolicyMetaclass(type):

denied/_sync/ability.py

@@ -0,0 +1,53 @@
+from typing import Any, Optional
+
+from denied.action import Action
+from denied.errors import UnauthorizedError, UndefinedPermission
+from denied.permission import Permission
+
+from .policy import Policy
+
+
+class Ability:
+    def __init__(
+        self, policy: Optional[Policy] = None, default_action: Action = Action.DENY
+    ):
+        self._policy = policy or Policy()
+        self._default_action = default_action
+
+    def authorize(self, permission: Permission, *args: Any, **kwargs: Any) -> None:
+        """Raises an UnauthorizedError if policy does not grant permission.
+
+        Args:
+            permission (Permission): a permission
+            args (Any): arguments passed to the policy access method
+            kwargs (Any): keyword argumentss passed to the policy access method
+
+        Returns:
+            None:
+        """
+        if not self.can(permission, *args, **kwargs):
+            raise UnauthorizedError(permission)
+
+    def can(self, permission: Permission, *args: Any, **kwargs: Any) -> bool:
+        """Returns the result of the policy access method defined for the permission.
+        If no access method is found the default_action is used.
+        If permission was not defined and default_action is RAISE then an
+        UndefinedPermission is raised.
+
+        Args:
+            permission (Permission): a permission
+            args (Any): arguments passed to the policy access method
+            kwargs (Any): keyword argumentss passed to the policy access method
+
+        Returns:
+            bool: True if permission is granted, False otherwise
+        """
+        try:
+            access_method = self._policy.get_access_method(permission)
+        except UndefinedPermission as error:
+            if self._default_action == Action.RAISE:
+                raise error
+            else:
+                return True if self._default_action == Action.ALLOW else False
+
+        return access_method(*args, **kwargs)

denied/_sync/policy.py

@@ -0,0 +1,77 @@
+from typing import Any, Callable, Dict, Optional, Tuple
+
+from denied.errors import PermissionAlreadyDefined, UndefinedPermission
+from denied.permission import Permission
+from denied.utils import SyncAccessMethod
+
+
+class PolicyMetaclass(type):
+    """Metaclass used by the Policy class.
+    It's used to register all the access methods defined by the @authorize() decorator.
+    """
+
+    def __new__(cls, name: str, bases: Tuple[type, ...], attrs: Dict[str, Any]) -> type:
+        """Callback called when a Policy class is created.
+        Loop over all the attributes and check if a `_authorized_permission` has
+        been defined on it.
+        If `_authorized_permission` is found we register the method
+        in a dictionary in order to access it later using the
+        permission being authorized.
+
+        Args:
+            cls: a Policy class
+            name (str): class name
+            bases (Tuple[type, ...]): base classes
+            attrs (Dict[str, Any]): class attributes
+        """
+        access_methods: Dict[Permission, str] = {}
+        for name, value in attrs.items():
+            permission: Optional[Permission] = getattr(
+                value, "_authorized_permission", None
+            )
+            if not permission:
+                continue
+
+            if permission in access_methods:
+                raise PermissionAlreadyDefined(permission)
+            access_methods[permission] = name
+
+        attrs["_access_methods"] = access_methods
+        return super().__new__(cls, name, bases, attrs)
+
+
+def authorize(permission: Permission) -> Callable[[SyncAccessMethod], SyncAccessMethod]:
+    def decorator(func: SyncAccessMethod) -> SyncAccessMethod:
+        """Add an `_authorized_permission` attribute to the method
+        in order for the metaclass to recognize it as an AccessMethod.
+
+        Args:
+            func (AccessMethod): method used to grant access
+
+        Returns:
+            AccessMethod: access method received as input
+        """
+        setattr(func, "_authorized_permission", permission)
+        return func
+
+    return decorator
+
+
+class Policy(metaclass=PolicyMetaclass):
+    _access_methods: Dict[Permission, str]
+
+    def get_access_method(self, permission: Permission) -> SyncAccessMethod:
+        """Returns the AccessMethod that was registered for the permission
+        received as input.
+        If no AccessMethod is found it raises a UndefinedPermission error.
+
+        Args:
+            permission (Permission): a permission
+
+        Returns:
+            AccessMethod: access method registered for permission
+        """
+        try:
+            return getattr(self, self._access_methods[permission])
+        except KeyError:
+            raise UndefinedPermission(permission)

denied/action.py

@@ -0,0 +1,7 @@
+from enum import Enum
+
+
+class Action(Enum):
+    DENY = "deny"
+    ALLOW = "allow"
+    RAISE = "raise"

denied/errors.py

@@ -27,3 +27,17 @@ def __init__(self, permission: Permission) -> None:
         """
         super().__init__(f"Access denied for permission {permission.name}")
         self.permission = permission
+
+
+class PermissionAlreadyDefined(Exception):
+    """Error raised while using the @authorize() decorator with the same Permission
+    twice in the same Policy object.
+    """
+
+    def __init__(self, permission: Permission) -> None:
+        """
+        Args:
+            permission (Permission): a permission
+        """
+        super().__init__(f"Permission {permission.name} already defined")
+        self.permission = permission

denied/ext/falcon.py

@@ -1,16 +1,19 @@
-from typing import Any, Callable, Optional
+from functools import wraps
+from typing import Any, Awaitable, Callable, Optional
 
 from falcon import Request
 from falcon.response import Response
 
-from denied.ability import Ability
+from denied import Ability
 from denied.ext.errors import AbilityNotFound
 from denied.permission import Permission
 
-ResourceMethod = Callable[[Any, Request, Any, Any], None]
+ResourceMethod = Callable[..., Awaitable[None]]
 
 
-def authorize(permission: Permission, ability_key: str = "ability"):
+def authorize(
+    permission: Permission, ability_key: str = "ability"
+) -> Callable[[ResourceMethod], ResourceMethod]:
     """Falcon's decorator for checking endpoints' permissions.
     The policy's access methods will be called with the request and
     all the other arguments and keyword arguments sent to the endpoint.
@@ -21,13 +24,14 @@ def authorize(permission: Permission, ability_key: str = "ability"):
     """
 
     def decorator(func: ResourceMethod) -> ResourceMethod:
-        def wrapper(
+        @wraps(func)
+        async def wrapper(
             resource: Any, req: Request, resp: Response, *args: Any, **kwargs: Any
         ) -> None:
             ability: Optional[Ability] = req.context.get(ability_key)
             if ability:
-                ability.authorize(permission, request=req, *args, **kwargs)
-                func(resource, req, resp, *args, **kwargs)
+                await ability.authorize(permission, request=req, *args, **kwargs)
+                await func(resource, req, resp, *args, **kwargs)
             else:
                 raise AbilityNotFound(
                     "Ability could not be found in "