IAM Role and Policy Compliance Checks

import boto3 from botocore.exceptions import ClientError, BotoCoreError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # Create a boto3 client for IAM iam_client = boto3.client('iam',aws_access_key_id=access_key,aws_secret_access_key=secret_key) try: # Retrieve the AWS account's details account_summary = iam_client.get_account_summary() # Check if any access keys exist for the root user root_access_keys = account_summary['SummaryMap'].get('AccountAccessKeysPresent', 0) if root_access_keys == 0: print("Compliant: No access keys found for the root user.") else: print("Non-compliant: Access keys found for the root user.") except ClientError as e: print(f"AWS client error occurred: {e}") except BotoCoreError as e: print(f"Boto core error occurred: {e}") except Exception as e: print(f"An unexpected error occurred: {e}")

import boto3 import botocore.exceptions creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # Initialize the IAM client iam_client = boto3.client('iam',aws_access_key_id=access_key,aws_secret_access_key=secret_key) try: # Create a paginator for the list_users operation paginator = iam_client.get_paginator('list_users') # Use the paginator to paginate through the users table = context.newtable() table.title = "User list" table.num_cols = 3 table.num_rows = 1 table.has_header_row = True rownum = 0 table.setval(rownum, 0, "User name") table.setval(rownum, 1, "User ID") table.setval(rownum, 2, "Created on") for page in paginator.paginate(): users = page['Users'] table.num_rows += len(page['Users']) # Output user details if users: # print("List of IAM Users:") for user in users: rownum += 1 # print(f"Username: {user['UserName']}, User ID: {user['UserId']}, Created On: {user['CreateDate']}") table.setval(rownum, 0, user['UserName']) table.setval(rownum, 1, user['UserId']) table.setval(rownum, 2, user['CreateDate']) else: print("No IAM users found in this page.") # Handle specific exceptions except botocore.exceptions.NoCredentialsError: print("Credentials not available") except botocore.exceptions.PartialCredentialsError: print("Incomplete credentials provided") except botocore.exceptions.SSLError: print("SSL connection could not be established. Ensure your network allows SSL connections to AWS services") except botocore.exceptions.EndpointConnectionError: print("Unable to connect to the endpoint. Check your AWS configuration and network settings") except botocore.exceptions.ClientError as e: print(f"Unexpected error occurred accessing AWS: {e}") # Handle general exceptions except Exception as e: print(f"An unhandled error occurred: {str(e)}")

import boto3 from botocore.exceptions import ClientError, BotoCoreError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def is_admin_policy(policy_document, exclude_permission_boundary): """ Check if the policy document contains admin access statements. """ for statement in policy_document.get('Statement', []): if statement.get('Effect') == 'Allow' and \ statement.get('Action') == '*' and \ statement.get('Resource') == '*': if not exclude_permission_boundary or \ (exclude_permission_boundary and 'Condition' not in statement): return True return False def evaluate_group_policies(user_name, iam, exclude_permission_boundary): """ Evaluate policies attached to groups for a given user. """ reasons = [] try: groups = iam.list_groups_for_user(UserName=user_name)['Groups'] for group in groups: attached_policies = iam.list_attached_group_policies(GroupName=group['GroupName'])['AttachedPolicies'] #print(attached_policies) # for debugging for policy in attached_policies: policy_details = iam.get_policy(PolicyArn=policy['PolicyArn']) if 'DefaultVersionId' in policy_details['Policy']: policy_version = iam.get_policy_version( PolicyArn=policy['PolicyArn'], VersionId=policy_details['Policy']['DefaultVersionId'] ) if is_admin_policy(policy_version['PolicyVersion']['Document'], exclude_permission_boundary): reasons.append(f"Group Attached Policy: {policy['PolicyArn']} ({group['GroupName']})") except ClientError as e: print(f"Error retrieving group policies for user {user_name}: {e}") return reasons def evaluate_attached_policies(user_name, iam, exclude_permission_boundary): """ Evaluate attached managed policies for a given user. """ reasons = [] try: attached_policies = iam.list_attached_user_policies(UserName=user_name)['AttachedPolicies'] #print(attached_policies) # for debugging for policy in attached_policies: policy_details = iam.get_policy(PolicyArn=policy['PolicyArn']) if 'DefaultVersionId' in policy_details['Policy']: policy_version = iam.get_policy_version( PolicyArn=policy['PolicyArn'], VersionId=policy_details['Policy']['DefaultVersionId'] ) if is_admin_policy(policy_version['PolicyVersion']['Document'], exclude_permission_boundary): reasons.append(f"Attached Policy: {policy['PolicyArn']}") except ClientError as e: print(f"Error retrieving attached policies for user {user_name}: {e}") return reasons def evaluate_inline_policies(user_name, iam, exclude_permission_boundary): """ Evaluate inline policies for a given user. """ reasons = [] try: inline_policies = iam.list_user_policies(UserName=user_name)['PolicyNames'] #print(inline_policies) # for debugging for policy_name in inline_policies: policy_document = iam.get_user_policy( UserName=user_name, PolicyName=policy_name )['PolicyDocument'] if is_admin_policy(policy_document, exclude_permission_boundary): reasons.append(f"Inline Policy: {policy_name}") except ClientError as e: print(f"Error retrieving inline policies for user {user_name}: {e}") return reasons def evaluate_iam_users_and_policies(exclude_permission_boundary=False): """ Evaluates IAM users for admin access in attached, inline, and group policies. """ iam = boto3.client('iam',aws_access_key_id=access_key,aws_secret_access_key=secret_key) compliance_report = { 'compliant': [], 'non_compliant': {} } try: users = iam.list_users()['Users'] for user in users: user_name = user['UserName'] print(f"Evaluating user: {user_name}") reasons = evaluate_attached_policies(user_name, iam, exclude_permission_boundary) + \ evaluate_inline_policies(user_name, iam, exclude_permission_boundary) + \ evaluate_group_policies(user_name, iam, exclude_permission_boundary) if reasons: compliance_report['non_compliant'][user_name] = reasons else: compliance_report['compliant'].append(user_name) except ClientError as e: print(f"ClientError while listing IAM users: {e}") except BotoCoreError as e: print(f"BotoCoreError encountered: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") # Print Compliance Report if compliance_report['non_compliant']: print("\nNon-Compliant IAM Users (Admin Access Found):") for user, reasons in compliance_report['non_compliant'].items(): print(f"{user} - Reasons: {', '.join(reasons)}") else: print("\nNo Non-Compliant IAM Users Found.") if compliance_report['compliant']: print("\nCompliant IAM Users (No Admin Access):") for user in compliance_report['compliant']: print(user) else: print("\nAll IAM Users are Non-Compliant.") exclude_permission_boundary = False evaluate_iam_users_and_policies(exclude_permission_boundary) context.skip_sub_tasks=True

import boto3 from botocore.exceptions import ClientError, BotoCoreError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def check_iam_users_for_attached_policies(): """ Check if any IAM users have policies attached directly to them. """ iam = boto3.client('iam',aws_access_key_id=access_key,aws_secret_access_key=secret_key) non_compliant_users = 0 total_users = 0 try: # Initialize pagination paginator = iam.get_paginator('list_users') for page in paginator.paginate(): for user in page['Users']: total_users += 1 username = user['UserName'] # Check for attached user policies attached_policies = iam.list_attached_user_policies(UserName=username)['AttachedPolicies'] if attached_policies: print(f"Non-compliant: User '{username}' has direct policies attached.") non_compliant_users += 1 else: print(f"Compliant: User '{username}' has no direct policies attached.") print(f"\nTotal users checked: {total_users}") print(f"Non-compliant users: {non_compliant_users}") except ClientError as e: print(f"AWS ClientError: {e.response['Error']['Message']}") except BotoCoreError as e: print(f"BotoCoreError: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") # Example usage check_iam_users_for_attached_policies() context.skip_sub_tasks=True

import boto3 import botocore.exceptions creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # Initialize the IAM client iam_client = boto3.client('iam',aws_access_key_id=access_key,aws_secret_access_key=secret_key) try: # Create a paginator for the list_users operation paginator = iam_client.get_paginator('list_users') # Use the paginator to paginate through the users table = context.newtable() table.title = "User list" table.num_cols = 3 table.num_rows = 1 table.has_header_row = True rownum = 0 table.setval(rownum, 0, "User name") table.setval(rownum, 1, "User ID") table.setval(rownum, 2, "Created on") for page in paginator.paginate(): users = page['Users'] table.num_rows += len(page['Users']) # Output user details if users: # print("List of IAM Users:") for user in users: rownum += 1 # print(f"Username: {user['UserName']}, User ID: {user['UserId']}, Created On: {user['CreateDate']}") table.setval(rownum, 0, user['UserName']) table.setval(rownum, 1, user['UserId']) table.setval(rownum, 2, user['CreateDate']) else: print("No IAM users found in this page.") # Handle specific exceptions except botocore.exceptions.NoCredentialsError: print("Credentials not available") except botocore.exceptions.PartialCredentialsError: print("Incomplete credentials provided") except botocore.exceptions.SSLError: print("SSL connection could not be established. Ensure your network allows SSL connections to AWS services") except botocore.exceptions.EndpointConnectionError: print("Unable to connect to the endpoint. Check your AWS configuration and network settings") except botocore.exceptions.ClientError as e: print(f"Unexpected error occurred accessing AWS: {e}") # Handle general exceptions except Exception as e: print(f"An unhandled error occurred: {str(e)}")

import boto3 from datetime import datetime creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # Initialize the IAM client iam_client = boto3.client('iam',aws_access_key_id=access_key,aws_secret_access_key=secret_key) # Define the threshold age in days #threshold_age = 55 # List to store old access key data old_keys_data = [] try: # Initialize a flag for old keys detection old_keys_found = False # Check access keys for each user for user in users: username = user['UserName'] access_keys = iam_client.list_access_keys(UserName=username)['AccessKeyMetadata'] # Check age of each access key for key in access_keys: #print(key) # for debugging key_age = (datetime.now(datetime.utcnow().astimezone().tzinfo) - key['CreateDate']).days if key_age > int(threshold_age): print(f"User: {username}, Access Key: {key['AccessKeyId']}, Age: {key_age} days, Status: {key['Status']}") old_keys_found = True # Add old key data to list old_keys_data.append({ 'username': username, 'access_key_id': key['AccessKeyId'] }) # If no old keys are found, print a message if not old_keys_found: print("No access keys older than the defined threshold were found.") else: print("Old key data: ", old_keys_data) # Pass `old_keys_data` to downstream task here except boto3.exceptions.botocore.exceptions.PartialCredentialsError as pce: print(f"Credentials error: {str(pce)}") except boto3.exceptions.botocore.exceptions.BotoCoreError as bce: print(f"BotoCore Error: {str(bce)}") except boto3.exceptions.botocore.exceptions.ClientError as ce: print(f"Client Error: {str(ce)}") except Exception as e: print(f"An unexpected error occurred: {str(e)}") context.skip_sub_tasks=True

import boto3 from botocore.exceptions import ClientError, BotoCoreError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def get_password_policy(client): """ Retrieves the current account password policy. :param client: Boto3 IAM client :return: Current password policy if exists, None otherwise """ try: return client.get_account_password_policy()['PasswordPolicy'] except client.exceptions.NoSuchEntityException: # No password policy is set for the account print("No password policy is set for the account.") return None except ClientError as e: print(f"Error retrieving password policy: {e}") return None def check_password_policy_compliance(current_policy, desired_policy): """ Checks if the current AWS IAM account password policy is compliant with the desired policy. Also, returns the non-compliant fields. :param current_policy: The current password policy :param desired_policy: The desired password policy attributes :return: Tuple (True if compliant, False otherwise, non_compliant_fields) """ non_compliant_fields = {} for key, value in desired_policy.items(): if key not in current_policy or current_policy[key] != value: non_compliant_fields[key] = { 'current_value': current_policy.get(key), 'desired_value': value } return len(non_compliant_fields) == 0, non_compliant_fields # Desired password policy parameters desired_policy = { 'MinimumPasswordLength': int(MinimumPasswordLength), 'RequireSymbols': RequireSymbols, 'RequireNumbers': RequireNumbers, 'RequireUppercaseCharacters': RequireUppercaseCharacters, 'RequireLowercaseCharacters': RequireLowercaseCharacters, 'MaxPasswordAge': int(MaxPasswordAge), # Days 'PasswordReusePrevention': int(PasswordReusePrevention), 'HardExpiry': HardExpiry } ''' # The 'HardExpiry' field in the password policy determines whether IAM users are allowed to change their own passwords. # - If 'HardExpiry' is set to True, it means IAM users cannot change their own passwords. In this case, only an administrator can reset the passwords. This setting is typically used in highly secure environments where password management needs to be strictly controlled by administrators. # - If 'HardExpiry' is set to False, IAM users are permitted to change their own passwords. This setting is more user-friendly and allows users to manage their own password changes, including regular updates or resets if needed. ''' # Create a boto3 client for IAM iam_client = boto3.client('iam',aws_access_key_id=access_key,aws_secret_access_key=secret_key) # Get the current account password policy current_policy = get_password_policy(iam_client) print("Current Policy:") for key, value in current_policy.items(): print(f" - {key}: {value}") if current_policy: # Debugging: Before checking compliance #print("Checking policy compliance...") is_compliant, non_compliant_fields = check_password_policy_compliance(current_policy, desired_policy) # Debugging: After checking compliance #print("Completed policy compliance check.") if is_compliant: print("The account password policy is compliant.") else: print("The account password policy is not compliant.") print("Non-compliant fields:") for field, values in non_compliant_fields.items(): print(f" - {field}: Current Value - {values['current_value']}, Desired Value - {values['desired_value']}") else: print("No password policy found for the account.") context.skip_sub_tasks=True #create_new_policy