WIP

Author: m-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,57 @@
-"""Centralization of all decision-making processes on a single decision point"""
+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.response import Err, Response
+from entitled.rules import Actor, Rule, RuleProto
 
 
 class Client:
-    "The Client class for decision-making centralization."
+    def __init__(self):
+        self._rule_registry: dict[str, Rule[Any]] = {}
 
-    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 define_rule(self, name: str, callable: RuleProto[Actor]) -> Rule[Actor]:
+        rule = Rule(name, callable)
+        self._rule_registry[rule.name] = rule
+        return rule
 
-    async def authorize(
+    async def inspect(
         self,
-        action: str,
-        actor: Any,
-        resource: Any,
-        context: dict[str, Any] | None = None,
-    ) -> bool:
-        policy = self._policy_lookup(resource)
-        return await policy.authorize(action, actor, resource, context)
+        name: str,
+        actor: Actor,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Response:
+        if name in self._rule_registry:
+            return await self._rule_registry[name].inspect(actor, *args, **kwargs)
+        return Err(f"No rule found with name '{name}'")
 
     async def allows(
         self,
-        action: str,
-        actor: Any,
-        resource: Any,
-        context: dict[str, Any] | None = None,
+        name: str,
+        actor: Actor,
+        *args: Any,
+        **kwargs: Any,
     ) -> bool:
-        policy = self._policy_lookup(resource)
-        return await policy.allows(action, actor, resource, context)
+        if name in self._rule_registry:
+            return await self._rule_registry[name].allows(actor, *args, **kwargs)
+        return False
 
-    async def grants(
+    async def denies(
         self,
-        actor: Any,
-        resource: Any,
-        context: dict[str, Any] | None = None,
-    ) -> dict[str, bool]:
-        policy = self._policy_lookup(resource)
-        return await 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")
+        name: str,
+        actor: Actor,
+        *args: Any,
+        **kwargs: Any,
+    ) -> bool:
+        return not self.allows(name, actor, *args, **kwargs)
 
-        return self._policy_registrar[lookup_key]
+    async def authorize(
+        self,
+        name: str,
+        actor: Actor,
+        *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}'")

entitled/old_client.py

@@ -0,0 +1,97 @@
+"""Centralization of all decision-making processes on a single decision point"""
+
+import importlib.util
+import pathlib
+import types
+from typing import Any
+
+from entitled import old_policies
+
+
+class Client:
+    "The Client class for decision-making centralization."
+
+    def __init__(self, base_path: str | None = None):
+        self._policy_registrar: dict[type, old_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)
+
+    async def authorize(
+        self,
+        action: str,
+        actor: Any,
+        resource: Any,
+        context: dict[str, Any] | None = None,
+    ) -> bool:
+        policy = self._policy_lookup(resource)
+        return await policy.authorize(action, actor, resource, context)
+
+    async def allows(
+        self,
+        action: str,
+        actor: Any,
+        resource: Any,
+        context: dict[str, Any] | None = None,
+    ) -> bool:
+        policy = self._policy_lookup(resource)
+        return await policy.allows(action, actor, resource, context)
+
+    async def grants(
+        self,
+        actor: Any,
+        resource: Any,
+        context: dict[str, Any] | None = None,
+    ) -> dict[str, bool]:
+        policy = self._policy_lookup(resource)
+        return await policy.grants(actor, resource, context)
+
+    def register(self, policy: old_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, old_policies.Policy):
+                try:
+                    self.register(attr)
+                except (ValueError, AttributeError):
+                    pass
+
+    def _policy_lookup(self, resource: Any) -> old_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]

entitled/old_policies.py

@@ -0,0 +1,89 @@
+"""Grouping of authorization rules around a particular resource type"""
+
+import asyncio
+
+from typing import Any, Callable, Generic, TypeVar
+
+from entitled import exceptions
+from entitled.old_rules import Rule, RuleProtocol
+
+T = TypeVar("T")
+
+
+class Policy(Generic[T]):
+    """A grouping of rules refering the given resource type."""
+
+    def __init__(
+        self,
+        label: str | None = None,
+        rules: dict[str, list[Rule[T]]] | None = None,
+    ):
+        self._registry: dict[
+            str,
+            list[Rule[T]],
+        ] = {}
+        self.label = label
+
+        if not rules:
+            rules = {}
+
+        for action, rule in rules.items():
+            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
+
+        return wrapped
+
+    def __register(self, action: str, *rules: Rule[T]):
+        if action not in self._registry:
+            self._registry[action] = [*rules]
+
+    async def grants(
+        self, actor: Any, resource: T | type[T], context: dict[str, Any] | None = None
+    ) -> dict[str, bool]:
+        return {
+            action: await self.allows(action, actor, resource, context)
+            for action in self._registry
+        }
+
+    async def allows(
+        self,
+        action: str,
+        actor: Any,
+        resource: T | type[T],
+        context: dict[str, Any] | None = None,
+    ) -> bool:
+        try:
+            return await self.authorize(action, actor, resource, context)
+        except exceptions.AuthorizationException:
+            return False
+
+    async def authorize(
+        self,
+        action: str,
+        actor: Any,
+        resource: T | type[T],
+        context: dict[str, Any] | None = None,
+    ) -> bool:
+        if action not in self._registry:
+            raise exceptions.UndefinedAction(
+                f"Action <{action}> undefined for this policy"
+            )
+
+        if not any(
+            (
+                await asyncio.gather(
+                    *(rule(actor, resource, context) for rule in self._registry[action])
+                )
+            )
+        ):
+            raise exceptions.AuthorizationException("Unauthorized")
+
+        return True