Merge pull request RedHatInsights#1152 from Ellen-Yi-Dong/log_delete_…

…group_role

Audit Logs: Log deletion of role and group

rbac/management/audit_log/model.py

@@ -69,21 +69,13 @@ def get_resource_item(self, r_type, request, *args, **kwargs):
         """Find related information (eg, name, id, etc...) for each resource item."""
         verify_tenant = self.get_tenant_id(request)
         if r_type == AuditLog.ROLE:
-            if request.data != {}:
-                role_object = get_object_or_404(Role, name=request.data["name"], tenant=verify_tenant)
-            else:
-                role_object = kwargs["kwargs"]
-            # retrieve role id and name
+            role_object = get_object_or_404(Role, name=request.data["name"], tenant=verify_tenant)
             role_object_id = role_object.id
             role_object_name = "role: " + role_object.name
             return role_object_id, role_object_name
 
         elif r_type == AuditLog.GROUP:
-            if request.data != {}:
-                group_object = get_object_or_404(Group, name=request.data["name"], tenant=verify_tenant)
-            else:
-                group_uuid = kwargs["kwargs"]["uuid"]
-                group_object = get_object_or_404(Group, uuid=group_uuid)
+            group_object = get_object_or_404(Group, name=request.data["name"], tenant=verify_tenant)
             group_object_id = group_object.id
             group_object_name = "group: " + group_object.name
             return group_object_id, group_object_name
@@ -92,15 +84,29 @@ def get_resource_item(self, r_type, request, *args, **kwargs):
             # TODO: update for permission related items
             return None
 
-    def log_create(self, request, resource, **kwargs):
+    def log_create(self, request, resource):
         """Audit Log when a role or a group is created."""
         self.principal_username = request.user.username
 
         self.resource_type = resource
 
-        self.resource_id, resource_name = self.get_resource_item(resource, request, kwargs=kwargs)
+        self.resource_id, resource_name = self.get_resource_item(resource, request)
         self.description = "Created " + resource_name
 
         self.action = AuditLog.CREATE
         self.tenant_id = self.get_tenant_id(request)
         super(AuditLog, self).save()
+
+    def log_delete(self, request, resource, object):
+        """Audit Log when a role or a group is deleted."""
+        self.principal_username = request.user.username
+
+        self.resource_type = resource
+        self.resource_id = object.id
+        resource_name = self.resource_type + ": " + object.name
+
+        self.description = "Deleted " + resource_name
+
+        self.action = AuditLog.DELETE
+        self.tenant_id = self.get_tenant_id(request)
+        super(AuditLog, self).save()

rbac/management/group/view.py

@@ -256,7 +256,7 @@ def create(self, request, *args, **kwargs):
 
         if status.is_success(create_group.status_code):
             auditlog = AuditLog()
-            auditlog.log_create(request, AuditLog.GROUP, kwargs=kwargs)
+            auditlog.log_create(request, AuditLog.GROUP)
 
         return create_group
 
@@ -368,6 +368,9 @@ def destroy(self, request, *args, **kwargs):
         response = super().destroy(request=request, args=args, kwargs=kwargs)
         if response.status_code == status.HTTP_204_NO_CONTENT:
             group_obj_change_notification_handler(request.user, group, "deleted")
+
+            auditlog = AuditLog()
+            auditlog.log_delete(request, AuditLog.GROUP, group)
         return response
 
     def update(self, request, *args, **kwargs):

rbac/management/role/view.py

@@ -214,7 +214,7 @@ def create(self, request, *args, **kwargs):
 
         if status.is_success(create_role.status_code):
             auditlog = AuditLog()
-            auditlog.log_create(request, AuditLog.ROLE, kwargs=kwargs)
+            auditlog.log_create(request, AuditLog.ROLE)
 
         return create_role
 
@@ -332,6 +332,9 @@ def destroy(self, request, *args, **kwargs):
             response = super().destroy(request=request, args=args, kwargs=kwargs)
         if response.status_code == status.HTTP_204_NO_CONTENT:
             role_obj_change_notification_handler(role, "deleted", request.user)
+
+            auditlog = AuditLog()
+            auditlog.log_delete(request, AuditLog.ROLE, role)
         return response
 
     def partial_update(self, request, *args, **kwargs):

tests/management/group/test_view.py

@@ -767,6 +767,24 @@ def test_delete_group_success(self, send_kafka_message):
 
             org_id = self.customer_data["org_id"]
 
+            # test whether correctly added to audit logs
+            al_url = "/api/v1/auditlogs/"
+            al_client = APIClient()
+            al_response = al_client.get(al_url, **self.headers)
+            retrieve_data = al_response.data.get("data")
+            al_list = retrieve_data
+            al_dict = al_list[0]
+
+            al_dict_principal_username = al_dict["principal_username"]
+            al_dict_description = al_dict["description"]
+            al_dict_resource = al_dict["resource_type"]
+            al_dict_action = al_dict["action"]
+
+            self.assertEqual(self.user_data["username"], al_dict_principal_username)
+            self.assertIsNotNone(al_dict_description)
+            self.assertEqual(al_dict_resource, "group")
+            self.assertEqual(al_dict_action, "delete")
+
             # verify the group no longer exists
             response = client.get(url, **self.headers)
             self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

tests/management/role/test_view.py

@@ -1327,6 +1327,23 @@ def test_delete_role_success(self, send_kafka_message):
             org_id = self.customer_data["org_id"]
 
             self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
+            # test whether correctly added to audit logs
+            al_url = "/api/v1/auditlogs/"
+            al_client = APIClient()
+            al_response = al_client.get(al_url, **self.headers)
+            retrieve_data = al_response.data.get("data")
+            al_list = retrieve_data
+            al_dict = al_list[1]
+
+            al_dict_principal_username = al_dict["principal_username"]
+            al_dict_description = al_dict["description"]
+            al_dict_resource = al_dict["resource_type"]
+            al_dict_action = al_dict["action"]
+
+            self.assertEqual(self.user_data["username"], al_dict_principal_username)
+            self.assertIsNotNone(al_dict_description)
+            self.assertEqual(al_dict_resource, "role")
+            self.assertEqual(al_dict_action, "delete")
 
             send_kafka_message.assert_called_with(
                 settings.NOTIFICATIONS_TOPIC,