Finishing up v1

Author: Mathias BIGAIGNON

entitled/client.py

@@ -1,3 +1,4 @@
+import re
 from typing import Any, Literal
 
 from entitled.exceptions import AuthorizationException
@@ -8,67 +9,77 @@
 
 class Client:
     def __init__(self):
-        self._policy_registry: dict[type, type[Policy[Any]]] = {}
+        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 _register_policy(self, policy: type[Policy[Any]]):
-        resource_type = getattr(policy, "__orig_bases__")[0].__args__[0]
+    def register_policy(self, policy: Policy[Any]):
+        resource_type = getattr(policy, "__orig_class__").__args__[0]
         self._policy_registry[resource_type] = policy
 
-    def _resolve_rule(self, name: str, resource: Any) -> Rule[Any] | None:
+    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)
-        if policy is not None:
-            rule = getattr(policy, name, None)
-            if rule is not None and hasattr(rule, "_is_rule"):
-                return Rule(name, rule)
-        return self._rule_registry.get(name, None)
+        return policy
 
     async def inspect(
         self,
         name: str,
-        actor: Actor,
+        actor: Any,
         resource: Any,
         *args: Any,
         **kwargs: Any,
     ) -> Response:
-        rule = self._resolve_rule(name, resource)
-        if rule is not None:
-            return await rule.inspect(actor, resource, *args, **kwargs)
-        return Err(f"No rule found with name '{name}'")
+        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,
         name: str,
-        actor: Actor,
+        actor: Any,
+        resource: Any,
         *args: Any,
         **kwargs: Any,
     ) -> bool:
-        if name in self._rule_registry:
-            return await self._rule_registry[name].allows(actor, *args, **kwargs)
-        return False
+        return (await self.inspect(name, actor, resource, *args, **kwargs)).allowed()
 
     async def denies(
         self,
         name: str,
-        actor: Actor,
+        actor: Any,
+        resource: Any,
         *args: Any,
         **kwargs: Any,
     ) -> bool:
-        return not self.allows(name, actor, *args, **kwargs)
+        return not await self.allows(name, actor, resource, *args, **kwargs)
 
     async def authorize(
         self,
         name: str,
-        actor: Actor,
+        actor: Any,
+        resource: Any,
         *args: Any,
         **kwargs: Any,
     ) -> Literal[True]:
-        if name in self._rule_registry:
-            return await self._rule_registry[name].authorize(actor, *args, **kwargs)
-        raise AuthorizationException(f"No rule found with name '{name}'")
+        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,
+        *args: Any,
+        **kwargs: Any,
+    ) -> dict[str, bool]:
+        policy = self._resolve_policy(resource)
+        if policy is None:
+            return {}
+        return await policy.grants(actor, resource, *args, **kwargs)

entitled/old_client.py

@@ -1,15 +1,93 @@
-import functools
-from typing import Any, Callable
+"""Grouping of authorization rules around a particular resource type"""
 
+from typing import Any, TypeVar
 
-def rule(func: Callable[..., Any]):
-    @functools.wraps(func)
-    async def wrapper(*args: Any, **kwargs: Any) -> Any:
-        return await func(*args, **kwargs)
+from entitled import exceptions
+from entitled.response import Err, Ok, Response
+from entitled.rules import Rule, RuleProto
 
-    setattr(wrapper, "_is_rule", True)
-    return wrapper
+T = TypeVar("T")
 
 
 class Policy[T]:
-    pass
+    """A grouping of rules refering the given resource type."""
+
+    _registry: dict[str, Rule[Any]]
+
+    def __init__(
+        self,
+        rules: dict[str, Rule[Any]] | None = None,
+    ):
+        self._registry = {}
+
+        if not rules:
+            rules = {}
+
+        for action, rule in rules.items():
+            self.register(action, *rule)
+
+    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
+
+    def register(self, action: str, rule: Rule[T]):
+        self._registry[action] = rule
+
+    async def inspect(
+        self,
+        action: str,
+        actor: Any,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Response:
+        if action not in self._registry:
+            return Err(f"Action <{action}> undefined for this policy")
+        return await self._registry[action].inspect(actor, *args, **kwargs)
+
+    async def allows(
+        self,
+        action: str,
+        actor: Any,
+        *args: Any,
+        **kwargs: Any,
+    ) -> bool:
+        return (await self.inspect(action, actor, *args, **kwargs)).allowed()
+
+    async def denies(
+        self,
+        action: str,
+        actor: Any,
+        *args: Any,
+        **kwargs: Any,
+    ) -> bool:
+        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
+        }