From 5fa58a55c05f006b81954bc571e7a9e1cca1ffed Mon Sep 17 00:00:00 2001 From: Kirill Sosnovskikh Date: Fri, 18 Oct 2024 13:25:12 +0300 Subject: [PATCH] [#304] Improve logging Boto3 IAM methods Signed-off-by: Kirill Sosnovskikh --- src/frostfs_testlib/s3/boto3_client.py | 174 +++++++++++++++++++------ 1 file changed, 135 insertions(+), 39 deletions(-) diff --git a/src/frostfs_testlib/s3/boto3_client.py b/src/frostfs_testlib/s3/boto3_client.py index b638939..a99b866 100644 --- a/src/frostfs_testlib/s3/boto3_client.py +++ b/src/frostfs_testlib/s3/boto3_client.py @@ -68,6 +68,7 @@ class Boto3ClientWrapper(S3ClientWrapper): self.access_key_id: str = access_key_id self.secret_access_key: str = secret_access_key self.s3gate_endpoint: str = "" + self.iam_endpoint: str = "" self.boto3_iam_client: S3Client = None self.set_endpoint(s3gate_endpoint) @@ -90,11 +91,16 @@ class Boto3ClientWrapper(S3ClientWrapper): @reporter.step("Set endpoint IAM to {iam_endpoint}") def set_iam_endpoint(self, iam_endpoint: str): + if self.iam_endpoint == iam_endpoint: + return + + self.iam_endpoint = iam_endpoint + self.boto3_iam_client = self.session.client( service_name="iam", aws_access_key_id=self.access_key_id, aws_secret_access_key=self.secret_access_key, - endpoint_url=iam_endpoint, + endpoint_url=self.iam_endpoint, verify=False, ) @@ -687,25 +693,36 @@ class Boto3ClientWrapper(S3ClientWrapper): # Some methods don't have checks because boto3 is silent in some cases (delete, attach, etc.) @reporter.step("Adds the specified user to the specified group") + @report_error def iam_add_user_to_group(self, user_name: str, group_name: str) -> dict: - response = self.boto3_iam_client.add_user_to_group(UserName=user_name, GroupName=group_name) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.add_user_to_group(**params) + log_command_execution(self.iam_endpoint, "IAM Add User to Group", response, params) return response @reporter.step("Attaches the specified managed policy to the specified IAM group") + @report_error def iam_attach_group_policy(self, group_name: str, policy_arn: str) -> dict: - response = self.boto3_iam_client.attach_group_policy(GroupName=group_name, PolicyArn=policy_arn) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.attach_group_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Attach Group Policy", response, params) sleep(S3_SYNC_WAIT_TIME * 10) return response @reporter.step("Attaches the specified managed policy to the specified user") + @report_error def iam_attach_user_policy(self, user_name: str, policy_arn: str) -> dict: - response = self.boto3_iam_client.attach_user_policy(UserName=user_name, PolicyArn=policy_arn) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.attach_user_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Attach User Policy", response, params) sleep(S3_SYNC_WAIT_TIME * 10) return response @reporter.step("Creates a new AWS secret access key and access key ID for the specified user") + @report_error def iam_create_access_key(self, user_name: str) -> dict: response = self.boto3_iam_client.create_access_key(UserName=user_name) + log_command_execution(self.iam_endpoint, "IAM Create Access Key", response, {"UserName": user_name}) access_key_id = response["AccessKey"].get("AccessKeyId") secret_access_key = response["AccessKey"].get("SecretAccessKey") @@ -715,138 +732,190 @@ class Boto3ClientWrapper(S3ClientWrapper): return access_key_id, secret_access_key @reporter.step("Creates a new group") + @report_error def iam_create_group(self, group_name: str) -> dict: response = self.boto3_iam_client.create_group(GroupName=group_name) + log_command_execution(self.iam_endpoint, "IAM Create Group", response, {"GroupName": group_name}) + assert response.get("Group"), f"Expected Group in response:\n{response}" assert response["Group"].get("GroupName") == group_name, f"GroupName should be equal to {group_name}" return response @reporter.step("Creates a new managed policy for your AWS account") + @report_error def iam_create_policy(self, policy_name: str, policy_document: dict) -> dict: - response = self.boto3_iam_client.create_policy(PolicyName=policy_name, PolicyDocument=json.dumps(policy_document)) + params = self._convert_to_s3_params(locals().items()) + params["PolicyDocument"] = json.dumps(policy_document) + response = self.boto3_iam_client.create_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Create Policy", response, params) + assert response.get("Policy"), f"Expected Policy in response:\n{response}" assert response["Policy"].get("PolicyName") == policy_name, f"PolicyName should be equal to {policy_name}" return response @reporter.step("Creates a new IAM user for your AWS account") + @report_error def iam_create_user(self, user_name: str) -> dict: response = self.boto3_iam_client.create_user(UserName=user_name) + log_command_execution(self.iam_endpoint, "IAM Create User", response, {"UserName": user_name}) + assert response.get("User"), f"Expected User in response:\n{response}" assert response["User"].get("UserName") == user_name, f"UserName should be equal to {user_name}" return response @reporter.step("Deletes the access key pair associated with the specified IAM user") + @report_error def iam_delete_access_key(self, access_key_id: str, user_name: str) -> dict: - response = self.boto3_iam_client.delete_access_key(AccessKeyId=access_key_id, UserName=user_name) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.delete_access_key(**params) + log_command_execution(self.iam_endpoint, "IAM Delete Access Key", response, params) return response @reporter.step("Deletes the specified IAM group") + @report_error def iam_delete_group(self, group_name: str) -> dict: response = self.boto3_iam_client.delete_group(GroupName=group_name) + log_command_execution(self.iam_endpoint, "IAM Delete Group", response, {"GroupName": group_name}) return response @reporter.step("Deletes the specified inline policy that is embedded in the specified IAM group") + @report_error def iam_delete_group_policy(self, group_name: str, policy_name: str) -> dict: - response = self.boto3_iam_client.delete_group_policy(GroupName=group_name, PolicyName=policy_name) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.delete_group_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Delete Group Policy", response, params) return response @reporter.step("Deletes the specified managed policy") + @report_error def iam_delete_policy(self, policy_arn: str) -> dict: response = self.boto3_iam_client.delete_policy(PolicyArn=policy_arn) + log_command_execution(self.iam_endpoint, "IAM Delete Policy", response, {"PolicyArn": policy_arn}) return response @reporter.step("Deletes the specified IAM user") + @report_error def iam_delete_user(self, user_name: str) -> dict: response = self.boto3_iam_client.delete_user(UserName=user_name) + log_command_execution(self.iam_endpoint, "IAM Delete User", response, {"UserName": user_name}) return response @reporter.step("Deletes the specified inline policy that is embedded in the specified IAM user") + @report_error def iam_delete_user_policy(self, user_name: str, policy_name: str) -> dict: - response = self.boto3_iam_client.delete_user_policy(UserName=user_name, PolicyName=policy_name) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.delete_user_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Delete User Policy", response, params) return response @reporter.step("Removes the specified managed policy from the specified IAM group") + @report_error def iam_detach_group_policy(self, group_name: str, policy_arn: str) -> dict: - response = self.boto3_iam_client.detach_group_policy(GroupName=group_name, PolicyArn=policy_arn) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.detach_group_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Detach Group Policy", response, params) sleep(S3_SYNC_WAIT_TIME * 10) return response @reporter.step("Removes the specified managed policy from the specified user") + @report_error def iam_detach_user_policy(self, user_name: str, policy_arn: str) -> dict: - response = self.boto3_iam_client.detach_user_policy(UserName=user_name, PolicyArn=policy_arn) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.detach_user_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Detach User Policy", response, params) sleep(S3_SYNC_WAIT_TIME * 10) return response @reporter.step("Returns a list of IAM users that are in the specified IAM group") + @report_error def iam_get_group(self, group_name: str) -> dict: response = self.boto3_iam_client.get_group(GroupName=group_name) + log_command_execution(self.iam_endpoint, "IAM Get Group", response, {"GroupName": group_name}) assert response.get("Group").get("GroupName") == group_name, f"GroupName should be equal to {group_name}" - return response @reporter.step("Retrieves the specified inline policy document that is embedded in the specified IAM group") + @report_error def iam_get_group_policy(self, group_name: str, policy_name: str) -> dict: - response = self.boto3_iam_client.get_group_policy(GroupName=group_name, PolicyName=policy_name) - + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.get_group_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Get Group Policy", response, params) return response @reporter.step("Retrieves information about the specified managed policy") + @report_error def iam_get_policy(self, policy_arn: str) -> dict: response = self.boto3_iam_client.get_policy(PolicyArn=policy_arn) + log_command_execution(self.iam_endpoint, "IAM Get Policy", response, {"PolicyArn": policy_arn}) + assert response.get("Policy"), f"Expected Policy in response:\n{response}" assert response["Policy"].get("Arn") == policy_arn, f"PolicyArn should be equal to {policy_arn}" return response @reporter.step("Retrieves information about the specified version of the specified managed policy") + @report_error def iam_get_policy_version(self, policy_arn: str, version_id: str) -> dict: - response = self.boto3_iam_client.get_policy_version(PolicyArn=policy_arn, VersionId=version_id) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.get_policy_version(**params) + log_command_execution(self.iam_endpoint, "IAM Get Policy Version", response, params) + assert response.get("PolicyVersion"), f"Expected PolicyVersion in response:\n{response}" assert response["PolicyVersion"].get("VersionId") == version_id, f"VersionId should be equal to {version_id}" return response @reporter.step("Retrieves information about the specified IAM user") + @report_error def iam_get_user(self, user_name: str) -> dict: response = self.boto3_iam_client.get_user(UserName=user_name) + log_command_execution(self.iam_endpoint, "IAM Get User", response, {"UserName": user_name}) + assert response.get("User"), f"Expected User in response:\n{response}" assert response["User"].get("UserName") == user_name, f"UserName should be equal to {user_name}" return response @reporter.step("Retrieves the specified inline policy document that is embedded in the specified IAM user") + @report_error def iam_get_user_policy(self, user_name: str, policy_name: str) -> dict: - response = self.boto3_iam_client.get_user_policy(UserName=user_name, PolicyName=policy_name) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.get_user_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Get User Policy", response, params) assert response.get("UserName"), f"Expected UserName in response:\n{response}" - return response @reporter.step("Returns information about the access key IDs associated with the specified IAM user") + @report_error def iam_list_access_keys(self, user_name: str) -> dict: response = self.boto3_iam_client.list_access_keys(UserName=user_name) - + log_command_execution(self.iam_endpoint, "IAM List Access Keys", response, {"UserName": user_name}) return response @reporter.step("Lists all managed policies that are attached to the specified IAM group") + @report_error def iam_list_attached_group_policies(self, group_name: str) -> dict: response = self.boto3_iam_client.list_attached_group_policies(GroupName=group_name) + log_command_execution(self.iam_endpoint, "IAM List Attached Group Policies", response, {"GroupName": group_name}) assert response.get("AttachedPolicies"), f"Expected AttachedPolicies in response:\n{response}" - return response @reporter.step("Lists all managed policies that are attached to the specified IAM user") + @report_error def iam_list_attached_user_policies(self, user_name: str) -> dict: response = self.boto3_iam_client.list_attached_user_policies(UserName=user_name) + log_command_execution(self.iam_endpoint, "IAM List Attached User Policies", response, {"UserName": user_name}) assert response.get("AttachedPolicies"), f"Expected AttachedPolicies in response:\n{response}" - return response @reporter.step("Lists all IAM users, groups, and roles that the specified managed policy is attached to") + @report_error def iam_list_entities_for_policy(self, policy_arn: str) -> dict: response = self.boto3_iam_client.list_entities_for_policy(PolicyArn=policy_arn) + log_command_execution(self.iam_endpoint, "IAM List Entities For Policy", response, {"PolicyArn": policy_arn}) assert response.get("PolicyGroups"), f"Expected PolicyGroups in response:\n{response}" assert response.get("PolicyUsers"), f"Expected PolicyUsers in response:\n{response}" @@ -854,98 +923,125 @@ class Boto3ClientWrapper(S3ClientWrapper): return response @reporter.step("Lists the names of the inline policies that are embedded in the specified IAM group") + @report_error def iam_list_group_policies(self, group_name: str) -> dict: response = self.boto3_iam_client.list_group_policies(GroupName=group_name) + log_command_execution(self.iam_endpoint, "IAM List Group Policies", response, {"GroupName": group_name}) assert response.get("PolicyNames"), f"Expected PolicyNames in response:\n{response}" - return response @reporter.step("Lists the IAM groups") + @report_error def iam_list_groups(self) -> dict: response = self.boto3_iam_client.list_groups() + log_command_execution(self.iam_endpoint, "IAM List Groups", response) assert response.get("Groups"), f"Expected Groups in response:\n{response}" - return response @reporter.step("Lists the IAM groups that the specified IAM user belongs to") + @report_error def iam_list_groups_for_user(self, user_name: str) -> dict: response = self.boto3_iam_client.list_groups_for_user(UserName=user_name) + log_command_execution(self.iam_endpoint, "IAM List Groups For User", response, {"UserName": user_name}) assert response.get("Groups"), f"Expected Groups in response:\n{response}" - return response @reporter.step("Lists all the managed policies that are available in your AWS account") + @report_error def iam_list_policies(self) -> dict: response = self.boto3_iam_client.list_policies() + log_command_execution(self.iam_endpoint, "IAM List Policies", response) assert response.get("Policies"), f"Expected Policies in response:\n{response}" - return response @reporter.step("Lists information about the versions of the specified managed policy") + @report_error def iam_list_policy_versions(self, policy_arn: str) -> dict: response = self.boto3_iam_client.list_policy_versions(PolicyArn=policy_arn) + log_command_execution(self.iam_endpoint, "IAM List Policy Versions", response, {"PolicyArn": policy_arn}) assert response.get("Versions"), f"Expected Versions in response:\n{response}" - return response @reporter.step("Lists the names of the inline policies embedded in the specified IAM user") + @report_error def iam_list_user_policies(self, user_name: str) -> dict: response = self.boto3_iam_client.list_user_policies(UserName=user_name) + log_command_execution(self.iam_endpoint, "IAM List User Policies", response, {"UserName": user_name}) assert response.get("PolicyNames"), f"Expected PolicyNames in response:\n{response}" - return response @reporter.step("Lists the IAM users") + @report_error def iam_list_users(self) -> dict: response = self.boto3_iam_client.list_users() + log_command_execution(self.iam_endpoint, "IAM List Users", response) assert response.get("Users"), f"Expected Users in response:\n{response}" - return response @reporter.step("Adds or updates an inline policy document that is embedded in the specified IAM group") + @report_error def iam_put_group_policy(self, group_name: str, policy_name: str, policy_document: dict) -> dict: - response = self.boto3_iam_client.put_group_policy( - GroupName=group_name, PolicyName=policy_name, PolicyDocument=json.dumps(policy_document) - ) + params = self._convert_to_s3_params(locals().items()) + params["PolicyDocument"] = json.dumps(policy_document) + response = self.boto3_iam_client.put_group_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Put Group Policy", response, params) sleep(S3_SYNC_WAIT_TIME * 10) return response @reporter.step("Adds or updates an inline policy document that is embedded in the specified IAM user") + @report_error def iam_put_user_policy(self, user_name: str, policy_name: str, policy_document: dict) -> dict: - response = self.boto3_iam_client.put_user_policy( - UserName=user_name, PolicyName=policy_name, PolicyDocument=json.dumps(policy_document) - ) + params = self._convert_to_s3_params(locals().items()) + params["PolicyDocument"] = json.dumps(policy_document) + response = self.boto3_iam_client.put_user_policy(**params) + log_command_execution(self.iam_endpoint, "IAM Put User Policy", response, params) sleep(S3_SYNC_WAIT_TIME * 10) return response @reporter.step("Removes the specified user from the specified group") + @report_error def iam_remove_user_from_group(self, group_name: str, user_name: str) -> dict: - response = self.boto3_iam_client.remove_user_from_group(GroupName=group_name, UserName=user_name) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.remove_user_from_group(**params) + log_command_execution(self.iam_endpoint, "IAM Remove User From Group", response, params) return response @reporter.step("Updates the name and/or the path of the specified IAM group") + @report_error def iam_update_group(self, group_name: str, new_name: str, new_path: Optional[str] = None) -> dict: - response = self.boto3_iam_client.update_group(GroupName=group_name, NewGroupName=new_name, NewPath="/") - + params = {"GroupName": group_name, "NewGroupName": new_name, "NewPath": "/"} + response = self.boto3_iam_client.update_group(**params) + log_command_execution(self.iam_endpoint, "IAM Update Group", response, params) return response @reporter.step("Updates the name and/or the path of the specified IAM user") + @report_error def iam_update_user(self, user_name: str, new_name: str, new_path: Optional[str] = None) -> dict: - response = self.boto3_iam_client.update_user(UserName=user_name, NewUserName=new_name, NewPath="/") + params = {"UserName": user_name, "NewUserName": new_name, "NewPath": "/"} + response = self.boto3_iam_client.update_user(**params) + log_command_execution(self.iam_endpoint, "IAM Update User", response, params) return response @reporter.step("Adds one or more tags to an IAM user") + @report_error def iam_tag_user(self, user_name: str, tags: list) -> dict: - tags_json = [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags] - response = self.boto3_iam_client.tag_user(UserName=user_name, Tags=tags_json) + params = self._convert_to_s3_params(locals().items()) + params["Tags"] = [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags] + response = self.boto3_iam_client.tag_user(**params) + log_command_execution(self.iam_endpoint, "IAM Tag User", response, params) return response @reporter.step("List tags of IAM user") + @report_error def iam_list_user_tags(self, user_name: str) -> dict: response = self.boto3_iam_client.list_user_tags(UserName=user_name) + log_command_execution(self.iam_endpoint, "IAM List User Tags", response, {"UserName": user_name}) return response @reporter.step("Removes the specified tags from the user") + @report_error def iam_untag_user(self, user_name: str, tag_keys: list) -> dict: - response = self.boto3_iam_client.untag_user(UserName=user_name, TagKeys=tag_keys) + params = self._convert_to_s3_params(locals().items()) + response = self.boto3_iam_client.untag_user(**params) + log_command_execution(self.iam_endpoint, "IAM Untag User", response, params) return response -- 2.45.2