Add .has_permission method and test of its usage wth server image

Author: tsmith023

.github/workflows/main.yaml

@@ -17,7 +17,7 @@ env:
   WEAVIATE_125: 1.25.24
   WEAVIATE_126: 1.26.8
   WEAVIATE_127: 1.27.1
-  WEAVIATE_128: 1.28.0-rc.0
+  WEAVIATE_128: 1.28.0-rc.0-d239d5c
 jobs:
   lint-and-format:
     name: Run Linter and Formatter

integration/test_rbac.py

@@ -301,5 +301,12 @@ def test_multiple_permissions(client_factory: ClientFactory) -> None:
             assert len(role.data_permissions) == 2
             assert role.data_permissions[0].action == Actions.Data.CREATE
             assert role.data_permissions[1].action == Actions.Data.UPDATE
+
+            assert client.roles.has_permission(
+                permission=role.collections_permissions[0], role=role_name
+            )
+            assert client.roles.has_permission(
+                permission=required_permissions[1][0], role=role_name
+            )
         finally:
             client.roles.delete(role_name)

weaviate/rbac/models.py

@@ -221,41 +221,91 @@ class CollectionsPermission:
     collection: str
     action: CollectionsAction
 
+    def _to_weaviate(self) -> WeaviatePermission:
+        return {
+            "action": self.action,
+            "collections": {
+                "collection": self.collection,
+                "tenant": "*",
+            },
+        }
+
 
 @dataclass
 class DataPermission:
     collection: str
     action: DataAction
 
+    def _to_weaviate(self) -> WeaviatePermission:
+        return {
+            "action": self.action,
+            "data": {
+                "collection": self.collection,
+                "object": "*",
+                "tenant": "*",
+            },
+        }
+
 
 @dataclass
 class RolesPermission:
     role: str
     action: RolesAction
 
+    def _to_weaviate(self) -> WeaviatePermission:
+        return {
+            "action": self.action,
+            "roles": {
+                "role": self.role,
+            },
+        }
+
 
 @dataclass
 class UsersPermission:
     action: UsersAction
 
+    def _to_weaviate(self) -> WeaviatePermission:
+        return {"action": self.action}
+
 
 @dataclass
 class ClusterPermission:
     action: ClusterAction
 
+    def _to_weaviate(self) -> WeaviatePermission:
+        return {"action": self.action}
+
 
 @dataclass
 class BackupsPermission:
     collection: str
     action: BackupsAction
 
+    def _to_weaviate(self) -> WeaviatePermission:
+        return {
+            "action": self.action,
+            "backups": {
+                "collection": self.collection,
+            },
+        }
+
 
 @dataclass
 class NodesPermission:
     collection: Optional[str]
     verbosity: Verbosity
     action: NodesAction
 
+    def _to_weaviate(self) -> WeaviatePermission:
+        return {
+            "action": self.action,
+            "nodes": {
+                "collection": self.collection or "*",
+                "verbosity": self.verbosity,
+            },
+        }
+
 
 PermissionsOutputType = Union[
     ClusterPermission,

weaviate/rbac/roles.py

@@ -5,6 +5,7 @@
 from weaviate.connect.v4 import _ExpectedStatusCodes
 from weaviate.rbac.models import (
     _Permission,
+    PermissionsOutputType,
     PermissionsInputType,
     Role,
     User,
@@ -129,6 +130,17 @@ async def _remove_permissions(self, permissions: List[WeaviatePermission], role:
             status_codes=_ExpectedStatusCodes(ok_in=[200], error="Remove permissions"),
         )
 
+    async def _has_permission(self, permission: WeaviatePermission, role: str) -> bool:
+        path = f"/authz/roles/{role}/has-permission"
+
+        res = await self._connection.post(
+            path,
+            weaviate_object=permission,
+            error_msg="Could not check permission",
+            status_codes=_ExpectedStatusCodes(ok_in=[200], error="Check permission"),
+        )
+        return cast(bool, res.json())
+
 
 class _RolesAsync(_RolesBase):
     def __user_from_weaviate_user(self, user: str) -> User:
@@ -279,6 +291,20 @@ async def remove_permissions(self, *, permissions: PermissionsInputType, role: s
             [permission._to_weaviate() for permission in _flatten_permissions(permissions)], role
         )
 
+    async def has_permission(
+        self, *, permission: Union[_Permission, PermissionsOutputType], role: str
+    ) -> bool:
+        """Check if a role has a specific permission.
+
+        Args:
+            permission: The permission to check.
+            role: The role to check the permission for.
+
+        Returns:
+            True if the role has the permission, False otherwise.
+        """
+        return await self._has_permission(permission._to_weaviate(), role)
+
 
 def _flatten_permissions(permissions: PermissionsInputType) -> List[_Permission]:
     if isinstance(permissions, _Permission):

weaviate/rbac/sync.pyi

@@ -1,6 +1,12 @@
 from typing import Dict, List, Optional, Union
 
-from weaviate.rbac.models import PermissionsInputType, Role, User
+from weaviate.rbac.models import (
+    _Permission,
+    PermissionsOutputType,
+    PermissionsInputType,
+    Role,
+    User,
+)
 from weaviate.rbac.roles import _RolesBase
 
 class _Roles(_RolesBase):
@@ -15,3 +21,6 @@ class _Roles(_RolesBase):
     def revoke(self, *, roles: Union[str, List[str]], user: str) -> None: ...
     def add_permissions(self, *, permissions: PermissionsInputType, role: str) -> None: ...
     def remove_permissions(self, *, permissions: PermissionsInputType, role: str) -> None: ...
+    def has_permission(
+        self, *, permission: Union[_Permission, PermissionsOutputType], role: str
+    ) -> bool: ...