Allow direct communication to Openshift RBAC API

Author: QuanMPhm

src/coldfront_plugin_cloud/management/commands/add_openshift_resource.py

@@ -7,12 +7,19 @@
     ResourceType,
 )
 
-from coldfront_plugin_cloud import attributes
+from coldfront_plugin_cloud import attributes, openshift
 
 
 class Command(BaseCommand):
     help = "Create OpenShift resource"
 
+    @staticmethod
+    def validate_role(role):
+        if role not in openshift.OPENSHIFT_ROLES:
+            raise ValueError(
+                f"Invalid role, {role} is not one of {', '.join(openshift.OPENSHIFT_ROLES)}"
+            )
+
     def add_arguments(self, parser):
         parser.add_argument(
             "--name", type=str, required=True, help="Name of OpenShift resource"
@@ -45,6 +52,8 @@ def add_arguments(self, parser):
         )
 
     def handle(self, *args, **options):
+        self.validate_role(options["role"])
+
         if options["for_virtualization"]:
             resource_description = "OpenShift Virtualization environment"
             resource_type = "OpenShift Virtualization"

src/coldfront_plugin_cloud/openshift.py

@@ -35,6 +35,9 @@
 }
 
 
+OPENSHIFT_ROLES = ["admin", "edit", "view"]
+
+
 def clean_openshift_metadata(obj):
     if "metadata" in obj:
         for attr in IGNORED_ATTRIBUTES:
@@ -253,25 +256,51 @@ def create_federated_user(self, unique_id):
         logger.info(f"User {unique_id} successfully created")
 
     def assign_role_on_user(self, username, project_id):
-        # /users/<user_name>/projects/<project>/roles/<role>
-        url = (
-            f"{self.auth_url}/users/{username}/projects/{project_id}"
-            f"/roles/{self.member_role_name}"
-        )
+        """Assign a role to a user in a project using direct OpenShift API calls"""
         try:
-            r = self.session.put(url)
-            self.check_response(r)
-        except Conflict:
+            # Try to get existing rolebindings with same name
+            # as the role name in project's namespace
+            rolebinding = self._openshift_get_rolebindings(
+                project_id, self.member_role_name
+            )
+
+            if not self._user_in_rolebinding(username, rolebinding):
+                # Add user to existing rolebinding
+                if "subjects" not in rolebinding:
+                    rolebinding["subjects"] = []
+                rolebinding["subjects"].append({"kind": "User", "name": username})
+                self._openshift_update_rolebindings(project_id, rolebinding)
+
+        except kexc.NotFoundError:
+            # Create new rolebinding if it doesn't exist
+            self._openshift_create_rolebindings(
+                project_id, username, self.member_role_name
+            )
+        except kexc.ConflictError:
+            # Role already exists, ignore
             pass
 
     def remove_role_from_user(self, username, project_id):
-        # /users/<user_name>/projects/<project>/roles/<role>
-        url = (
-            f"{self.auth_url}/users/{username}/projects/{project_id}"
-            f"/roles/{self.member_role_name}"
-        )
-        r = self.session.delete(url)
-        self.check_response(r)
+        """Remove a role from a user in a project using direct OpenShift API calls"""
+        try:
+            rolebinding = self._openshift_get_rolebindings(
+                project_id, self.member_role_name
+            )
+
+            if "subjects" in rolebinding:
+                rolebinding["subjects"] = [
+                    subject
+                    for subject in rolebinding["subjects"]
+                    if not (
+                        subject.get("kind") == "User"
+                        and subject.get("name") == username
+                    )
+                ]
+                self._openshift_update_rolebindings(project_id, rolebinding)
+
+        except kexc.NotFoundError:
+            # Rolebinding doesn't exist, nothing to remove
+            pass
 
     def _create_project(self, project_name, project_id):
         url = f"{self.auth_url}/projects/{project_id}"
@@ -290,13 +319,13 @@ def _create_project(self, project_name, project_id):
         self.check_response(r)
 
     def _get_role(self, username, project_id):
-        # /users/<user_name>/projects/<project>/roles/<role>
-        url = (
-            f"{self.auth_url}/users/{username}/projects/{project_id}"
-            f"/roles/{self.member_role_name}"
+        rolebindings = self._openshift_get_rolebindings(
+            project_id, self.member_role_name
         )
-        r = self.session.get(url)
-        return self.check_response(r)
+        if not self._user_in_rolebinding(username, rolebindings):
+            raise NotFound(
+                f"User {username} has no rolebindings in project {project_id}"
+            )
 
     def _get_project(self, project_id):
         url = f"{self.auth_url}/projects/{project_id}"
@@ -308,10 +337,24 @@ def _delete_user(self, username):
         self._openshift_delete_identity(username)
         logger.info(f"User {username} successfully deleted")
 
-    def get_users(self, project_id):  # TODO (Quan) This should also replaced
-        url = f"{self.auth_url}/projects/{project_id}/users"
-        r = self.session.get(url)
-        return set(self.check_response(r))
+    def get_users(self, project_id):
+        """Get all users with roles in a project"""
+        users = set()
+
+        # Check all standard OpenShift roles
+        for role in OPENSHIFT_ROLES:
+            try:
+                rolebinding = self._openshift_get_rolebindings(project_id, role)
+                if "subjects" in rolebinding:
+                    users.update(
+                        subject["name"]
+                        for subject in rolebinding["subjects"]
+                        if subject.get("kind") == "User"
+                    )
+            except kexc.NotFoundError:
+                continue
+
+        return users
 
     def _openshift_get_user(self, username):
         api = self.get_resource_api(API_USER, "User")
@@ -446,3 +489,51 @@ def _openshift_delete_resourcequota(self, project_id, resourcequota_name):
         """In an openshift namespace {project_id) delete a specified resourcequota"""
         api = self.get_resource_api(API_CORE, "ResourceQuota")
         return api.delete(namespace=project_id, name=resourcequota_name).to_dict()
+
+    def _user_in_rolebinding(self, username, rolebinding):
+        """Check if a user is in a rolebinding"""
+        if "subjects" not in rolebinding:
+            return False
+
+        return any(
+            subject.get("kind") == "User" and subject.get("name") == username
+            for subject in rolebinding["subjects"]
+        )
+
+    def _openshift_get_rolebindings(self, project_name, role):
+        api = self.get_resource_api(API_RBAC, "RoleBinding")
+        result = clean_openshift_metadata(
+            api.get(namespace=project_name, name=role).to_dict()
+        )
+
+        # Ensure subjects is a list
+        if not result.get("subjects"):
+            result["subjects"] = []
+
+        return result
+
+    def _openshift_create_rolebindings(self, project_name, username, role):
+        api = self.get_resource_api(API_RBAC, "RoleBinding")
+        payload = {
+            "metadata": {"name": role, "namespace": project_name},
+            "subjects": [{"name": username, "kind": "User"}],
+            "roleRef": {"name": role, "kind": "ClusterRole"},
+        }
+        return clean_openshift_metadata(
+            api.create(body=payload, namespace=project_name).to_dict()
+        )
+
+    def _openshift_update_rolebindings(self, project_name, rolebinding):
+        api = self.get_resource_api(API_RBAC, "RoleBinding")
+        return clean_openshift_metadata(
+            api.patch(body=rolebinding, namespace=project_name).to_dict()
+        )
+
+    def _openshift_list_rolebindings(self, project_name):
+        """List all rolebindings in a project"""
+        api = self.get_resource_api(API_RBAC, "RoleBinding")
+        try:
+            result = clean_openshift_metadata(api.get(namespace=project_name).to_dict())
+            return result.get("items", [])
+        except kexc.NotFoundError:
+            return []

src/coldfront_plugin_cloud/tests/unit/openshift/test_rbac.py

@@ -0,0 +1,134 @@
+from unittest import mock
+
+import kubernetes.dynamic.exceptions as kexc
+
+from coldfront_plugin_cloud.tests import base
+from coldfront_plugin_cloud.openshift import OpenShiftResourceAllocator
+
+
+class TestMocOpenShiftRBAC(base.TestBase):
+    def setUp(self) -> None:
+        mock_resource = mock.Mock()
+        mock_allocation = mock.Mock()
+        self.allocator = OpenShiftResourceAllocator(mock_resource, mock_allocation)
+        self.allocator.id_provider = "fake_idp"
+        self.allocator.k8_client = mock.Mock()
+        self.allocator.member_role_name = "admin"
+
+    def test_user_in_rolebindings_false(self):
+        fake_rb = {
+            "subjects": [
+                {
+                    "kind": "User",
+                    "name": "fake-user-2",
+                }
+            ]
+        }
+        output = self.allocator._user_in_rolebinding("fake-user", fake_rb)
+        self.assertFalse(output)
+
+    def test_user_in_rolebindings_true(self):
+        fake_rb = {
+            "subjects": [
+                {
+                    "kind": "User",
+                    "name": "fake-user",
+                }
+            ]
+        }
+        output = self.allocator._user_in_rolebinding("fake-user", fake_rb)
+        self.assertTrue(output)
+
+    @mock.patch(
+        "coldfront_plugin_cloud.openshift.OpenShiftResourceAllocator._openshift_get_rolebindings"
+    )
+    @mock.patch(
+        "coldfront_plugin_cloud.openshift.OpenShiftResourceAllocator._openshift_update_rolebindings"
+    )
+    def test_add_user_to_role(self, fake_update_rb, fake_get_rb):
+        fake_get_rb.return_value = {
+            "subjects": [],
+        }
+        self.allocator.assign_role_on_user("fake-user", "fake-project")
+        fake_update_rb.assert_called_with(
+            "fake-project", {"subjects": [{"kind": "User", "name": "fake-user"}]}
+        )
+
+    @mock.patch(
+        "coldfront_plugin_cloud.openshift.OpenShiftResourceAllocator._openshift_get_rolebindings"
+    )
+    @mock.patch(
+        "coldfront_plugin_cloud.openshift.OpenShiftResourceAllocator._openshift_create_rolebindings"
+    )
+    def test_add_user_to_role_not_exists(self, fake_create_rb, fake_get_rb):
+        fake_error = kexc.NotFoundError(mock.Mock())
+        fake_get_rb.side_effect = fake_error
+        self.allocator.assign_role_on_user("fake-user", "fake-project")
+        fake_create_rb.assert_called_with("fake-project", "fake-user", "admin")
+
+    @mock.patch(
+        "coldfront_plugin_cloud.openshift.OpenShiftResourceAllocator._openshift_get_rolebindings"
+    )
+    def test_remove_user_from_role(self, fake_get_rb):
+        fake_get_rb.return_value = {
+            "subjects": [{"kind": "User", "name": "fake-user"}],
+        }
+        self.allocator.k8_client.resources.get.return_value.patch.return_value.to_dict.return_value = {}
+        self.allocator.remove_role_from_user("fake-user", "fake-project")
+        self.allocator.k8_client.resources.get.return_value.patch.assert_called_with(
+            body={"subjects": []}, namespace="fake-project"
+        )
+
+    def test_remove_user_from_role_not_exists(self):
+        fake_error = kexc.NotFoundError(mock.Mock())
+        self.allocator.k8_client.resources.get.return_value.get.side_effect = fake_error
+        self.allocator.remove_role_from_user("fake-project", "fake-user")
+        self.allocator.k8_client.resources.get.return_value.patch.assert_not_called()
+
+    def test_get_rolebindings(self):
+        fake_rb = mock.Mock(spec=["to_dict"])
+        fake_rb.to_dict.return_value = {"subjects": []}
+        self.allocator.k8_client.resources.get.return_value.get.return_value = fake_rb
+        res = self.allocator._openshift_get_rolebindings("fake-project", "admin")
+        self.assertEqual(res, fake_rb.to_dict())
+
+    def test_get_rolebindings_no_subjects(self):
+        fake_rb = mock.Mock(spec=["to_dict"])
+        fake_rb.to_dict.return_value = {}
+        self.allocator.k8_client.resources.get.return_value.get.return_value = fake_rb
+        res = self.allocator._openshift_get_rolebindings("fake-project", "admin")
+        self.assertEqual(res, {"subjects": []})
+
+    def test_list_rolebindings(self):
+        fake_rb = mock.Mock(spec=["to_dict"])
+        fake_rb.to_dict.return_value = {
+            "items": ["rb1", "rb2"],
+        }
+        self.allocator.k8_client.resources.get.return_value.get.return_value = fake_rb
+        res = self.allocator._openshift_list_rolebindings("fake-project")
+        self.assertEqual(res, ["rb1", "rb2"])
+
+    def test_list_rolebindings_not_exists(self):
+        fake_error = kexc.NotFoundError(mock.Mock())
+        self.allocator.k8_client.resources.get.return_value.get.side_effect = fake_error
+        res = self.allocator._openshift_list_rolebindings("fake-project")
+        self.assertEqual(res, [])
+
+    def test_create_rolebindings(self):
+        fake_rb = mock.Mock(spec=["to_dict"])
+        fake_rb.to_dict.return_value = {}
+        self.allocator.k8_client.resources.get.return_value.create.return_value = (
+            fake_rb
+        )
+        res = self.allocator._openshift_create_rolebindings(
+            "fake-project", "fake-user", "admin"
+        )
+        self.assertEqual(res, {})
+        self.allocator.k8_client.resources.get.return_value.create.assert_called_with(
+            namespace="fake-project",
+            body={
+                "metadata": {"name": "admin", "namespace": "fake-project"},
+                "subjects": [{"name": "fake-user", "kind": "User"}],
+                "roleRef": {"name": "admin", "kind": "ClusterRole"},
+            },
+        )