Squash merge v1.0 into main

Author: Mathias BIGAIGNON

entitled/__init__.py

@@ -1,7 +0,0 @@
-"""Importing some base classes at package root level for convenience"""
-
-from entitled.client import Client
-from entitled.policies import Policy
-from entitled.rules import Rule
-
-__all__ = ["Rule", "Policy", "Client"]

entitled/client.py

@@ -1,97 +1,85 @@
-"""Centralization of all decision-making processes on a single decision point"""
+import re
+from typing import Any, Literal
 
-import importlib.util
-import pathlib
-import types
-from typing import Any
-
-from entitled import policies
+from entitled.exceptions import AuthorizationException
+from entitled.policies import Policy
+from entitled.response import Err, Response
+from entitled.rules import Actor, Rule, RuleProto
 
 
 class Client:
-    "The Client class for decision-making centralization."
+    def __init__(self):
+        self._policy_registry: dict[type, Policy[Any]] = {}
+        self._rule_registry: dict[str, Rule[Any]] = {}
+
+    def define_rule(self, name: str, callable: RuleProto[Actor]) -> Rule[Actor]:
+        rule = Rule(name, callable)
+        self._rule_registry[rule.name] = rule
+        return rule
 
-    def __init__(self, base_path: str | None = None):
-        self._policy_registrar: dict[type, policies.Policy[Any]] = {}
-        self._load_path = None
-        if base_path:
-            self._load_path = pathlib.Path(base_path)
-            self.load_policies_from_path(self._load_path)
+    def register_policy(self, policy: Policy[Any]):
+        resource_type = getattr(policy, "__orig_class__").__args__[0]
+        self._policy_registry[resource_type] = policy
 
-    def authorize(
+    def _resolve_policy(self, resource: Any) -> Policy[Any] | None:
+        lookup_key = resource if isinstance(resource, type) else type(resource)
+        policy = self._policy_registry.get(lookup_key, None)
+        return policy
+
+    async def inspect(
+        self,
+        name: str,
+        actor: Any,
+        resource: Any,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Response:
+        policy = self._resolve_policy(resource)
+        if policy is not None:
+            return await policy.inspect(name, actor, resource, *args, **kwargs)
+        return Err(f"No policy found with name '{name}'")
+
+    async def allows(
         self,
-        action: str,
+        name: str,
         actor: Any,
         resource: Any,
-        context: dict[str, Any] | None = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> bool:
-        policy = self._policy_lookup(resource)
-        return policy.authorize(action, actor, resource, context)
+        return (await self.inspect(name, actor, resource, *args, **kwargs)).allowed()
 
-    def allows(
+    async def denies(
         self,
-        action: str,
+        name: str,
         actor: Any,
         resource: Any,
-        context: dict[str, Any] | None = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> bool:
-        policy = self._policy_lookup(resource)
-        return policy.allows(action, actor, resource, context)
+        return not await self.allows(name, actor, resource, *args, **kwargs)
 
-    def grants(
+    async def authorize(
+        self,
+        name: str,
+        actor: Any,
+        resource: Any,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Literal[True]:
+        result = await self.inspect(name, actor, resource, *args, **kwargs)
+        if not result.allowed():
+            raise AuthorizationException(result.message())
+        return True
+
+    async def grants(
         self,
         actor: Any,
         resource: Any,
-        context: dict[str, Any] | None = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> dict[str, bool]:
-        policy = self._policy_lookup(resource)
-        return policy.grants(actor, resource, context)
-
-    def register(self, policy: policies.Policy[Any]):
-        if hasattr(policy, "__orig_class__"):
-            resource_type = getattr(policy, "__orig_class__").__args__[0]
-            if resource_type not in self._policy_registrar:
-                self._policy_registrar[resource_type] = policy
-            else:
-                raise ValueError(
-                    "A policy is already registered for this resource type"
-                )
-        else:
-            raise AttributeError(f"Policy {policy} is incorrectly defined")
-
-    def reload_registrar(self):
-        if self._load_path is not None:
-            self.load_policies_from_path(self._load_path)
-
-    def load_policies_from_path(self, path: pathlib.Path):
-        for file_path in path.glob("*.py"):
-            print(file_path)
-            mod_name = file_path.stem
-            full_module_name = ".".join(file_path.parts[:-1] + (mod_name,))
-            spec = importlib.util.spec_from_file_location(full_module_name, file_path)
-            if spec:
-                module = importlib.util.module_from_spec(spec)
-                if spec.loader:
-                    try:
-                        spec.loader.exec_module(module)
-                    except Exception as e:
-                        raise e
-
-                    self._register_from_module(module)
-
-    def _register_from_module(self, module: types.ModuleType):
-        for attribute_name in dir(module):
-            attr = getattr(module, attribute_name)
-            if isinstance(attr, policies.Policy):
-                try:
-                    self.register(attr)
-                except (ValueError, AttributeError):
-                    pass
-
-    def _policy_lookup(self, resource: Any) -> policies.Policy[Any]:
-        lookup_key = resource if isinstance(resource, type) else type(resource)
-
-        if lookup_key not in self._policy_registrar:
-            raise ValueError("No policy registered for this resource type")
-
-        return self._policy_registrar[lookup_key]
+        policy = self._resolve_policy(resource)
+        if policy is None:
+            return {}
+        return await policy.grants(actor, resource, *args, **kwargs)

entitled/policies.py

@@ -1,81 +1,93 @@
 """Grouping of authorization rules around a particular resource type"""
 
-from typing import Any, Callable, Generic, TypeVar
+from typing import Any, TypeVar
 
 from entitled import exceptions
-from entitled.rules import Rule, RuleProtocol
+from entitled.response import Err, Ok, Response
+from entitled.rules import Rule, RuleProto
 
 T = TypeVar("T")
 
 
-class Policy(Generic[T]):
+class Policy[T]:
     """A grouping of rules refering the given resource type."""
 
+    _registry: dict[str, Rule[Any]]
+
     def __init__(
         self,
-        label: str | None = None,
-        rules: dict[str, list[Rule[T]]] | None = None,
+        rules: dict[str, Rule[Any]] | None = None,
     ):
-        self._registry: dict[
-            str,
-            list[Rule[T]],
-        ] = {}
-        self.label = label
+        self._registry = {}
 
         if not rules:
             rules = {}
 
         for action, rule in rules.items():
-            self.__register(action, *rule)
+            self.register(action, *rule)
 
-    def rule(self, name: str) -> Callable[[RuleProtocol[T]], RuleProtocol[T]]:
-        def wrapped(func: RuleProtocol[T]):
-            rule_name = name
-            if self.label is not None:
-                rule_name = self.label + ":" + rule_name
-            new_rule = Rule[T](rule_name, func)
-            self.__register(name, new_rule)
-            return func
+    def rule(self, func: RuleProto[Any]):
+        rule_name = f"{func.__name__}"
+        new_rule = Rule[T](rule_name, func)
+        self.register(rule_name, new_rule)
+        return func
 
-        return wrapped
+    def register(self, action: str, rule: Rule[T]):
+        self._registry[action] = rule
 
-    def __register(self, action: str, *rules: Rule[T]):
+    async def inspect(
+        self,
+        action: str,
+        actor: Any,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Response:
         if action not in self._registry:
-            self._registry[action] = [*rules]
-
-    def grants(
-        self, actor: Any, resource: T | type[T], context: dict[str, Any] | None = None
-    ) -> dict[str, bool]:
-        return {
-            action: self.allows(action, actor, resource, context)
-            for action in self._registry
-        }
+            return Err(f"Action <{action}> undefined for this policy")
+        return await self._registry[action].inspect(actor, *args, **kwargs)
 
-    def allows(
+    async def allows(
         self,
         action: str,
         actor: Any,
-        resource: T | type[T],
-        context: dict[str, Any] | None = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> bool:
-        try:
-            return self.authorize(action, actor, resource, context)
-        except exceptions.AuthorizationException:
-            return False
+        return (await self.inspect(action, actor, *args, **kwargs)).allowed()
 
-    def authorize(
+    async def denies(
         self,
         action: str,
         actor: Any,
-        resource: T | type[T],
-        context: dict[str, Any] | None = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> bool:
-        if action not in self._registry:
-            raise exceptions.UndefinedAction(
-                f"Action <{action}> undefined for this policy"
-            )
-
-        if not any(rule(actor, resource, context) for rule in self._registry[action]):
-            raise exceptions.AuthorizationException("Unauthorized")
+        return not (await self.inspect(action, actor, *args, **kwargs)).allowed()
 
+    async def authorize(
+        self,
+        action: str,
+        actor: Any,
+        *args: Any,
+        **kwargs: Any,
+    ) -> bool:
+        res = await self.inspect(action, actor, *args, **kwargs)
+        if not res.allowed():
+            raise exceptions.AuthorizationException(res.message())
         return True
+
+    async def grants(
+        self,
+        actor: Any,
+        *args: Any,
+        **kwargs: Any,
+    ) -> dict[str, bool]:
+        return {
+            action: await self.allows(
+                action,
+                actor,
+                *args,
+                **kwargs,
+            )
+            for action in self._registry
+        }

entitled/response.py

@@ -0,0 +1,27 @@
+from dataclasses import dataclass
+from typing import TypeVar
+
+T = TypeVar("T")
+
+
+@dataclass(frozen=True)
+class Ok:
+    def allowed(self):
+        return True
+
+    def message(self):
+        return None
+
+
+@dataclass(frozen=True)
+class Err:
+    msg: str = ""
+
+    def allowed(self):
+        return False
+
+    def message(self):
+        return self.msg
+
+
+Response = Ok | Err