Migrate to Pydantic

py_abac/_policy/__init__.py

@@ -0,0 +1,5 @@
+"""
+    Exposed classes and methods
+"""
+
+# from .policy import Policy

py_abac/_policy/conditions/__init__.py

@@ -0,0 +1,3 @@
+"""
+    Conditions
+"""

py_abac/_policy/conditions/attribute/__init__.py

@@ -0,0 +1,12 @@
+"""
+    Attribute conditions
+"""
+
+from .all_in import AllInAttribute
+from .all_not_in import AllNotInAttribute
+from .any_in import AnyInAttribute
+from .any_not_in import AnyNotInAttribute
+from .equals import EqualsAttribute
+from .is_in import IsInAttribute
+from .is_not_in import IsNotInAttribute
+from .not_equals import NotEqualsAttribute

py_abac/_policy/conditions/attribute/all_in.py

@@ -0,0 +1,46 @@
+"""
+    All in attribute condition
+"""
+
+import logging
+
+from .base import AttributeCondition
+from ..collection.base import is_collection
+
+LOG = logging.getLogger(__name__)
+
+
+class AllInAttribute(AttributeCondition):
+    """
+        Condition for all attribute values in that of another
+    """
+    # Condition type specifier
+    condition: str = "AllInAttribute"
+
+    def is_satisfied(self, ctx) -> bool:
+        # Extract attribute value from request to match
+        self._value = ctx.get_attribute_value(self.ace, self.path)
+        # Check if attribute value to match is a collection
+        if not is_collection(ctx.attribute_value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(ctx.attribute_value),
+                ctx.attribute_path,
+                ctx.ace
+            )
+            return False
+        return self._is_satisfied(ctx.attribute_value)
+
+    def _is_satisfied(self, what) -> bool:
+        # Check if value is a collection
+        if not is_collection(self._value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(self._value),
+                self.path,
+                self.ace
+            )
+            return False
+        return set(what).issubset(self._value)

py_abac/_policy/conditions/attribute/all_not_in.py

@@ -0,0 +1,46 @@
+"""
+    All not in attribute condition
+"""
+
+import logging
+
+from .base import AttributeCondition
+from ..collection.base import is_collection
+
+LOG = logging.getLogger(__name__)
+
+
+class AllNotInAttribute(AttributeCondition):
+    """
+        Condition for all attribute values not in that of another
+    """
+    # Condition type specifier
+    condition: str = "AllNotInAttribute"
+
+    def is_satisfied(self, ctx) -> bool:
+        # Extract attribute value from request to match
+        self._value = ctx.get_attribute_value(self.ace, self.path)
+        # Check if attribute value to match is a collection
+        if not is_collection(ctx.attribute_value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(ctx.attribute_value),
+                ctx.attribute_path,
+                ctx.ace
+            )
+            return False
+        return self._is_satisfied(ctx.attribute_value)
+
+    def _is_satisfied(self, what) -> bool:
+        # Check if value is a collection
+        if not is_collection(self._value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(self._value),
+                self.path,
+                self.ace
+            )
+            return False
+        return not set(what).issubset(self._value)

py_abac/_policy/conditions/attribute/any_in.py

@@ -0,0 +1,46 @@
+"""
+    Any in attribute condition
+"""
+
+import logging
+
+from .base import AttributeCondition
+from ..collection.base import is_collection
+
+LOG = logging.getLogger(__name__)
+
+
+class AnyInAttribute(AttributeCondition):
+    """
+        Condition for any attribute values in that of another
+    """
+    # Condition type specifier
+    condition: str = "AnyInAttribute"
+
+    def is_satisfied(self, ctx) -> bool:
+        # Extract attribute value from request to match
+        self._value = ctx.get_attribute_value(self.ace, self.path)
+        # Check if attribute value to match is a collection
+        if not is_collection(ctx.attribute_value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(ctx.attribute_value),
+                ctx.attribute_path,
+                ctx.ace
+            )
+            return False
+        return self._is_satisfied(ctx.attribute_value)
+
+    def _is_satisfied(self, what) -> bool:
+        # Check if value is a collection
+        if not is_collection(self._value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(self._value),
+                self.path,
+                self.ace
+            )
+            return False
+        return bool(set(what).intersection(self._value))

py_abac/_policy/conditions/attribute/any_not_in.py

@@ -0,0 +1,46 @@
+"""
+    Any not in attribute condition
+"""
+
+import logging
+
+from .base import AttributeCondition
+from ..collection.base import is_collection
+
+LOG = logging.getLogger(__name__)
+
+
+class AnyNotInAttribute(AttributeCondition):
+    """
+        Condition for any attribute values not in that of another
+    """
+    # Condition type specifier
+    condition: str = "AnyNotInAttribute"
+
+    def is_satisfied(self, ctx) -> bool:
+        # Extract attribute value from request to match
+        self._value = ctx.get_attribute_value(self.ace, self.path)
+        # Check if attribute value to match is a collection
+        if not is_collection(ctx.attribute_value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(ctx.attribute_value),
+                ctx.attribute_path,
+                ctx.ace
+            )
+            return False
+        return self._is_satisfied(ctx.attribute_value)
+
+    def _is_satisfied(self, what) -> bool:
+        # Check if value is a collection
+        if not is_collection(self._value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(self._value),
+                self.path,
+                self.ace
+            )
+            return False
+        return not bool(set(what).intersection(self._value))

py_abac/_policy/conditions/attribute/base.py

@@ -0,0 +1,86 @@
+"""
+    Attribute Condition Base
+"""
+
+from typing import Any
+
+from marshmallow import ValidationError
+from objectpath import Tree
+from pydantic import PrivateAttr
+
+from ..base import ConditionBase, ABCMeta, abstractmethod
+
+
+def validate_path(path):
+    """
+        Validate given attribute path satisfies ObjectPath notation.
+        Throws ValidationError for invalid path.
+    """
+    try:
+        Tree({}).execute(path)
+    except Exception as err:
+        raise ValidationError(*err.args)
+
+
+class AccessControlElementField(str):
+    """
+        Access control element field
+    """
+    # Access control elements
+    aces = ["subject", "resource", "action", "context"]
+
+    @classmethod
+    def __get_validators__(cls):
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, v):
+        if v not in cls.aces:
+            raise ValueError("Must be one of: {}".format(cls.aces))
+        return v
+
+
+class ObjectPathField(str):
+    """
+        ObjectPath field
+    """
+
+    @classmethod
+    def __get_validators__(cls):
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, v):
+        try:
+            Tree({}).execute(v)
+        except Exception:
+            raise ValueError("Invalid ObjectPath notation")
+        return v
+
+
+class AttributeCondition(ConditionBase, metaclass=ABCMeta):
+    """
+        Base class for attribute conditions
+    """
+    ace: AccessControlElementField
+    path: ObjectPathField
+    _value: Any = PrivateAttr()
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._value = None
+
+    def is_satisfied(self, ctx) -> bool:
+        # Extract attribute value from request to match
+        self._value = ctx.get_attribute_value(self.ace, self.path)
+        return self._is_satisfied(ctx.attribute_value)
+
+    @abstractmethod
+    def _is_satisfied(self, what) -> bool:
+        """
+            Is attribute conditions satisfied
+
+            :param what: attribute value to check
+            :return: True if satisfied else False
+        """
+        raise NotImplementedError()

py_abac/_policy/conditions/attribute/equals.py

@@ -0,0 +1,16 @@
+"""
+    Equals attribute condition
+"""
+
+from .base import AttributeCondition
+
+
+class EqualsAttribute(AttributeCondition):
+    """
+        Condition for attribute value equals that of another
+    """
+    # Condition type specifier
+    condition: str = "EqualsAttribute"
+
+    def _is_satisfied(self, what) -> bool:
+        return what == self._value

py_abac/_policy/conditions/attribute/is_in.py

@@ -0,0 +1,31 @@
+"""
+    Is in attribute condition
+"""
+
+import logging
+
+from .base import AttributeCondition
+from ..collection.base import is_collection
+
+LOG = logging.getLogger(__name__)
+
+
+class IsInAttribute(AttributeCondition):
+    """
+        Condition for attribute value in that of another
+    """
+    # Condition type specifier
+    condition: str = "IsInAttribute"
+
+    def _is_satisfied(self, what) -> bool:
+        # Check if value is a collection
+        if not is_collection(self._value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(self._value),
+                self.path,
+                self.ace
+            )
+            return False
+        return what in self._value

py_abac/_policy/conditions/attribute/is_not_in.py

@@ -0,0 +1,31 @@
+"""
+    Is not in attribute condition
+"""
+
+import logging
+
+from .base import AttributeCondition
+from ..collection.base import is_collection
+
+LOG = logging.getLogger(__name__)
+
+
+class IsNotInAttribute(AttributeCondition):
+    """
+        Condition for attribute value not in that of another
+    """
+    # Condition type specifier
+    condition: str = "IsNotInAttribute"
+
+    def _is_satisfied(self, what) -> bool:
+        # Check if value is a collection
+        if not is_collection(self._value):
+            LOG.debug(
+                "Invalid type '%s' for attribute value at path '%s' for element '%s'."
+                " Condition not satisfied.",
+                type(self._value),
+                self.path,
+                self.ace
+            )
+            return False
+        return what not in self._value

py_abac/_policy/conditions/attribute/not_equals.py

@@ -0,0 +1,16 @@
+"""
+    Not equals attribute condition
+"""
+
+from .base import AttributeCondition
+
+
+class NotEqualsAttribute(AttributeCondition):
+    """
+        Condition for attribute value not equals that of another
+    """
+    # Condition type specifier
+    condition: str = "NotEqualsAttribute"
+
+    def _is_satisfied(self, what) -> bool:
+        return what != self._value