RBAC:Python

# -*- coding: utf-8 -*-

from typing import Any, Dict

from recc.rbac.rbac_domain import RbacDomain
from recc.rbac.rbac_exception import RbacKeyError
from recc.rbac.rbac_permission import RbacPermission
from recc.rbac.rbac_role import RbacRole
from recc.rbac.rbac_subject import RbacSubject


class Rbac:
    def __init__(self):
        self._domains: Dict[Any, RbacDomain] = dict()
        self._roles: Dict[Any, RbacRole] = dict()
        self._permissions: Dict[Any, RbacPermission] = dict()
        self._subjects: Dict[Any, RbacSubject] = dict()

    def create_domain(self, key: Any) -> RbacDomain:
        if key in self._domains:
            raise RbacKeyError(f"Already exists domain-key: {key}")
        domain = RbacDomain(self, key)
        self._domains[key] = domain
        return domain

    def create_role(self, key: Any) -> RbacRole:
        if key in self._roles:
            raise RbacKeyError(f"Already exists role-key: {key}")
        role = RbacRole(self, key)
        self._roles[key] = role
        return role

    def create_permission(self, key: Any) -> RbacPermission:
        if key in self._permissions:
            raise RbacKeyError(f"Already exists permission-key: {key}")
        perm = RbacPermission(self, key)
        self._permissions[key] = perm
        return perm

    def create_subject(self, key: Any) -> RbacSubject:
        if key in self._subjects:
            raise RbacKeyError(f"Already exists subject-key: {key}")
        subject = RbacSubject(self, key)
        self._subjects[key] = subject
        return subject

# -*- coding: utf-8 -*-

from typing import Any, Dict, Optional

from recc.rbac.rbac_exception import RbacKeyError, RbacMismatchContextError
from recc.rbac.rbac_permission import RbacPermission
from recc.rbac.rbac_role import RbacRole
from recc.rbac.rbac_subject import RbacSubject


class RbacDomain:
    def __init__(self, rbac, key: Any, parent: Optional["RbacDomain"] = None):
        self._rbac = rbac
        self._key = key
        self._parent = parent
        self._subdomains: Dict[Any, RbacDomain] = dict()
        self._subjects: Dict[RbacSubject, RbacRole] = dict()

    def __repr__(self) -> str:
        return f"RbacDomain[rbac={id(self._rbac)},key={self._key}]"

    def __str__(self) -> str:
        return self._key

    @property
    def context(self):
        return self._rbac

    def create_subdomain(self, key: Any) -> "RbacDomain":
        if key in self._subdomains:
            raise RbacKeyError(f"Already exists subdomain-key: {key}")
        project = RbacDomain(self._rbac, key, self)
        self._subdomains[key] = project
        return project

    def add_subject(self, subject: RbacSubject, role: RbacRole) -> None:
        if self.context != subject.context:
            raise RbacMismatchContextError
        if self.context != role.context:
            raise RbacMismatchContextError
        self._subjects[subject] = role

    def test_permission(
        self,
        subject: RbacSubject,
        *permission: RbacPermission,
        inherit=True,
    ) -> None:
        if self.context != subject.context:
            raise RbacMismatchContextError
        for p in permission:
            if self.context != p.context:
                raise RbacMismatchContextError

        if subject in self._subjects:
            role = self._subjects[subject]
            for p in permission:
                role.test_permission(p)
        elif inherit and self._parent is not None:
            self._parent.test_permission(subject, *permission, inherit=inherit)
        else:
            raise RbacKeyError(f"Not found subject: {str(subject)}")

    def has_permission(
        self,
        subject: RbacSubject,
        *permissions: RbacPermission,
        inherit=True,
    ) -> bool:
        try:
            self.test_permission(subject, *permissions, inherit=inherit)
        except:  # noqa
            return False
        else:
            return True

# -*- coding: utf-8 -*-


class RbacError(Exception):
    def __init__(self, *args):
        super().__init__(*args)


class RbacKeyError(RbacError):
    def __init__(self, *args):
        super().__init__(*args)


class RbacMismatchContextError(RbacError):
    def __init__(self, *args):
        super().__init__(*args)


class RbacPermissionError(RbacError):
    def __init__(self, *args):
        super().__init__(*args)

# -*- coding: utf-8 -*-

from typing import Any


class RbacPermission:
    def __init__(self, rbac, key: Any):
        self._rbac = rbac
        self._key = key

    def __repr__(self) -> str:
        return f"RbacPermission[rbac={id(self._rbac)},key={self._key}]"

    def __str__(self) -> str:
        return self._key

    @property
    def context(self):
        return self._rbac

    @property
    def key(self) -> Any:
        return self._key

# -*- coding: utf-8 -*-

from typing import Any, Set

from recc.rbac.rbac_exception import RbacMismatchContextError, RbacPermissionError
from recc.rbac.rbac_permission import RbacPermission


class RbacRole:
    def __init__(self, rbac, key: Any):
        self._rbac = rbac
        self._key = key
        self._allows: Set[RbacPermission] = set()
        self._denies: Set[RbacPermission] = set()

    def __repr__(self) -> str:
        return f"RbacRole[rbac={id(self._rbac)},key={self._key}]"

    def __str__(self) -> str:
        return self._key

    @property
    def context(self):
        return self._rbac

    @property
    def key(self) -> Any:
        return self._key

    def allow_permission(self, *permissions: RbacPermission) -> None:
        for p in permissions:
            if self.context != p.context:
                raise RbacMismatchContextError

        for p in permissions:
            self._allows.add(p)

    def deny_permission(self, *permissions: RbacPermission) -> None:
        for p in permissions:
            if self.context != p.context:
                raise RbacMismatchContextError

        for p in permissions:
            self._denies.add(p)

    def test_allow_permission(self, permission: RbacPermission) -> None:
        if self.context != permission.context:
            raise RbacMismatchContextError

        if permission not in self._allows:
            raise RbacPermissionError(f"Not allowed permission: {str(permission)}")

    def test_deny_permission(self, permission: RbacPermission) -> None:
        if self.context != permission.context:
            raise RbacMismatchContextError

        if permission in self._denies:
            raise RbacPermissionError(f"Denied permission: {str(permission)}")

    def test_permission(self, permission: RbacPermission) -> None:
        if self.context != permission.context:
            raise RbacMismatchContextError

        self.test_allow_permission(permission)
        self.test_deny_permission(permission)

# -*- coding: utf-8 -*-

from typing import Any


class RbacSubject:
    def __init__(self, rbac, key: Any):
        self._rbac = rbac
        self._key = key

    def __repr__(self) -> str:
        return f"RbacSubject[rbac={id(self._rbac)},key={self._key}]"

    def __str__(self) -> str:
        return self._key

    @property
    def context(self):
        return self._rbac

    @property
    def key(self) -> Any:
        return self._key

# -*- coding: utf-8 -*-

from unittest import TestCase, main

from recc.rbac.rbac import Rbac
from recc.rbac.rbac_exception import RbacMismatchContextError


class RbacTestCase(TestCase):
    def test_basic_scenario(self):
        rbac = Rbac()

        # Permissions
        view = rbac.create_permission("view")
        edit = rbac.create_permission("edit")

        # Roles
        admin = rbac.create_role("admin")
        guest = rbac.create_role("guest")

        # Role to Permission
        admin.allow_permission(edit, view)
        guest.allow_permission(view)

        # Users
        user1 = rbac.create_subject("user1")
        user2 = rbac.create_subject("user2")
        user3 = rbac.create_subject("user3")

        # Groups
        group1 = rbac.create_domain("group1")

        # Projects
        project1 = group1.create_subdomain("project1")
        project2 = group1.create_subdomain("project2")

        # Group members
        group1.add_subject(user1, admin)
        group1.add_subject(user2, guest)
        self.assertTrue(group1.has_permission(user1, view))
        self.assertTrue(group1.has_permission(user1, edit))
        self.assertTrue(group1.has_permission(user2, view))
        self.assertFalse(group1.has_permission(user2, edit))
        self.assertFalse(group1.has_permission(user3, view))
        self.assertFalse(group1.has_permission(user3, edit))

        # Project members
        project1.add_subject(user1, guest)
        project1.add_subject(user2, admin)
        self.assertTrue(project1.has_permission(user1, view))
        self.assertFalse(project1.has_permission(user1, edit))
        self.assertTrue(project1.has_permission(user2, view))
        self.assertTrue(project1.has_permission(user2, edit))
        self.assertFalse(project1.has_permission(user3, view))
        self.assertFalse(project1.has_permission(user3, edit))

        project2.add_subject(user3, guest)
        self.assertTrue(project2.has_permission(user1, view))
        self.assertTrue(project2.has_permission(user1, edit))
        self.assertTrue(project2.has_permission(user2, view))
        self.assertFalse(project2.has_permission(user2, edit))
        self.assertTrue(project2.has_permission(user3, view))
        self.assertFalse(project2.has_permission(user3, edit))

        self.assertFalse(project2.has_permission(user1, view, inherit=False))
        self.assertFalse(project2.has_permission(user1, edit, inherit=False))
        self.assertFalse(project2.has_permission(user2, view, inherit=False))
        self.assertFalse(project2.has_permission(user2, edit, inherit=False))
        self.assertTrue(project2.has_permission(user3, view, inherit=False))
        self.assertFalse(project2.has_permission(user3, edit, inherit=False))

    def test_context_mismatch(self):
        rbac1 = Rbac()
        view1 = rbac1.create_permission("view")
        role1 = rbac1.create_role("role")

        rbac2 = Rbac()
        view2 = rbac2.create_permission("view")
        role2 = rbac2.create_role("role")

        with self.assertRaises(RbacMismatchContextError):
            role2.allow_permission(view1)
        with self.assertRaises(RbacMismatchContextError):
            role1.allow_permission(view2)


if __name__ == "__main__":
    main()