Allow direct communication to Openshift RBAC API

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:
@@ -232,33 +235,72 @@ def get_federated_user(self, username):
         logger.info(f"User ({username}) does not exist")
 
     def create_federated_user(self, unique_id):
-        url = f"{self.auth_url}/users/{unique_id}"
-        try:
-            r = self.session.put(url)
-            self.check_response(r)
-        except Conflict:
-            pass
+        user_def = {
+            "metadata": {"name": unique_id},
+            "fullName": unique_id,
+        }
+
+        identity_def = {
+            "providerName": self.id_provider,
+            "providerUserName": unique_id,
+        }
+
+        identity_mapping_def = {
+            "user": {"name": unique_id},
+            "identity": {"name": self.qualified_id_user(unique_id)},
+        }
+
+        self._openshift_create_user(user_def)
+        self._openshift_create_identity(identity_def)
+        self._openshift_create_useridentitymapping(identity_mapping_def)
+        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}"
@@ -277,39 +319,84 @@ 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}"
         r = self.session.get(url)
         return self.check_response(r)
 
     def _delete_user(self, username):
-        url = f"{self.auth_url}/users/{username}"
-        r = self.session.delete(url)
-        return self.check_response(r)
+        self._openshift_delete_user(username)
+        self._openshift_delete_identity(username)
+        logger.info(f"User {username} successfully deleted")
 
     def get_users(self, project_id):
-        url = f"{self.auth_url}/projects/{project_id}/users"
-        r = self.session.get(url)
-        return set(self.check_response(r))
+        """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")
         return clean_openshift_metadata(api.get(name=username).to_dict())
 
+    def _openshift_create_user(self, user_def):
+        api = self.get_resource_api(API_USER, "User")
+        try:
+            return clean_openshift_metadata(api.create(body=user_def).to_dict())
+        except kexc.ConflictError:
+            pass
+
+    def _openshift_delete_user(self, username):
+        api = self.get_resource_api(API_USER, "User")
+        return clean_openshift_metadata(api.delete(name=username).to_dict())
+
     def _openshift_get_identity(self, id_user):
         api = self.get_resource_api(API_USER, "Identity")
         return clean_openshift_metadata(
             api.get(name=self.qualified_id_user(id_user)).to_dict()
         )
 
+    def _openshift_create_identity(self, identity_def):
+        api = self.get_resource_api(API_USER, "Identity")
+        try:
+            return clean_openshift_metadata(api.create(body=identity_def).to_dict())
+        except kexc.ConflictError:
+            pass
+
+    def _openshift_delete_identity(self, username):
+        api = self.get_resource_api(API_USER, "Identity")
+        return api.delete(name=self.qualified_id_user(username)).to_dict()
+
+    def _openshift_create_useridentitymapping(self, identity_mapping_def):
+        api = self.get_resource_api(API_USER, "UserIdentityMapping")
+        try:
+            return clean_openshift_metadata(
+                api.create(body=identity_mapping_def).to_dict()
+            )
+        except kexc.ConflictError:
+            pass
+
     def _openshift_user_exists(self, user_name):
         try:
             self._openshift_get_user(user_name)
@@ -402,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/functional/openshift/test_allocation.py

@@ -286,3 +286,33 @@ def test_project_default_labels(self):
         self.assertTrue(
             namespace_dict_labels.items() > openshift.PROJECT_DEFAULT_LABELS.items()
         )
+
+    def test_create_incomplete(self):
+        """Creating a user that only has user, but no identity or mapping should not raise an error."""
+        user = self.new_user()
+        project = self.new_project(pi=user)
+        allocation = self.new_allocation(project, self.resource, 1)
+        allocator = openshift.OpenShiftResourceAllocator(self.resource, allocation)
+        user_def = {
+            "metadata": {"name": user.username},
+            "fullName": user.username,
+        }
+
+        allocator._openshift_create_user(user_def)
+        self.assertTrue(allocator._openshift_user_exists(user.username))
+        self.assertFalse(allocator._openshift_identity_exists(user.username))
+        self.assertFalse(
+            allocator._openshift_useridentitymapping_exists(
+                user.username, user.username
+            )
+        )
+
+        # Now create identity and mapping, no errors should be raised
+        allocator.get_or_create_federated_user(user.username)
+        self.assertTrue(allocator._openshift_user_exists(user.username))
+        self.assertTrue(allocator._openshift_identity_exists(user.username))
+        self.assertTrue(
+            allocator._openshift_useridentitymapping_exists(
+                user.username, user.username
+            )
+        )