import logging from datetime import datetime, timezone from typing import Any, Dict import boto3 from botocore.exceptions import ClientError from jinja2 import Environment, PackageLoader, select_autoescape from user_access_revoker.const import APP_NAME from user_access_revoker.utils import days_diff, get_tag_value_by_key env = Environment(loader=PackageLoader(APP_NAME), autoescape=select_autoescape()) email_subject = env.get_template("email_subject.txt") email_body_txt = env.get_template("email_body.txt") email_body_html = env.get_template("email_body.html") def get_tags_for_user(client: boto3.session.Session.client, username: str) -> list: """return list of tags of the user""" response = client.list_user_tags(UserName=username) return response["Tags"] if "Tags" in response else [] def get_user_access_keys(client: boto3.session.Session.client, username: str) -> list: """get user's access keys""" response = client.list_access_keys(UserName=username) return response["AccessKeyMetadata"] if "AccessKeyMetadata" in response else [] def get_all_users(client: boto3.session.Session.client): """get list of all users""" users = [] paginator = client.get_paginator("list_users") page_iterator = paginator.paginate() for page in page_iterator: for user in page.get("Users"): username = user.get("UserName") print(username) tags = get_tags_for_user(client, username) users.append( { "UserName": username, "Email": get_tag_value_by_key(tags, "Email"), "BackupEmail": get_tag_value_by_key(tags, "backup_email"), "FullName": get_tag_value_by_key(tags, "Name"), "CreateDate": user.get("CreateDate"), "PasswordLastUsed": user.get("PasswordLastUsed", datetime.today()), "Tags": tags, "AccessKeys": get_user_access_keys(client, username), "WatchDogEnabled": get_tag_value_by_key(tags, "watchdog_enabled").lower() == "always", "NotificationEnabled": get_tag_value_by_key(tags, "notification_enabled").lower() == "always", } ) return users def is_key_in_use(client: boto3.session.Session.client, key_id: str) -> bool: """Check whether access key was used at all""" last_used = client.get_access_key_last_used(AccessKeyId=key_id) if "LastUsedDate" in last_used["AccessKeyLastUsed"]: return True else: return False def email_user( client: boto3.session.Session.client, logger: logging.Logger, recipient: str = "WRONG E-MAIL ADDR", subject: str = "AWS Watchdog dummy letter", body_html: str = "No text", body_text: str = "No text", charset: str = "UTF-8", sender: str = "AWS Watchdog ", senderArn: str = "arn:aws:ses:us-east-1:483193324480:identity/aws-user-alerts.delphiplatform.io" ): """Notify AWS User By e-mail Args: client: boto3.session.Session.client instance logger: Logger instance recipient: E-mail (Default: WRONG E-MAIL ADDR) subject: E-mail subject (Default: AWS Watchdog dummy letter ) body_html: E-mail html body (Default: No text ) body_text: E-mail text body (Default: No text ) charset: E-mail Charset (Default: utf8 ) sender: E-mail FROM (Default: AWS Watchdog ) senderArn: ARN of identity """ # Try to send the email. try: response = client.send_email( Destination={ "ToAddresses": [ recipient, ], }, Message={ "Body": { "Html": { "Charset": charset, "Data": body_html, }, "Text": { "Charset": charset, "Data": body_text, }, }, "Subject": { "Charset": charset, "Data": subject, }, }, Source=sender, SourceArn=senderArn, ReplyToAddresses=["devops@sonymusic.com"], ) except ClientError as e: error_response = e.response["Error"]["Message"] logger.error(f"Email sent to {recipient} failed - {error_response}.") else: message_id = response["MessageId"] logger.info(f"Email sent to {recipient} MessageId {message_id}.") return message_id def handler(lambda_context, logger: logging.Logger, log_context: Dict[str, Any], config, payload: Any) -> Any: """Handler contains business logic Args: lambda_context: Contains information about runtime logger: Logger instance log_context: Put additional parameters here and put to log as "extra" parameter config: Application config payload: Data you want to process Returns: JSON serializable response """ iam_client = boto3.client("iam") ses_client = boto3.client("ses") is_dry_run = config.DRY_RUN aws_account_alias = iam_client.list_account_aliases().get("AccountAliases")[0] current_date = datetime.now(timezone.utc) for user in get_all_users(iam_client): is_notification_enabled = user.get("NotificationEnabled") if user.get("WatchDogEnabled"): for key_info in user.get("AccessKeys"): status = key_info.get("Status") access_key_id = key_info.get("AccessKeyId") access_key_id_masked = f"{access_key_id[:5]}**********{access_key_id[15:]}" created_date = key_info.get("CreateDate") if status != "Inactive": key_age = days_diff(current_date, created_date) logger.debug( f"{user.get('UserName')} has key {access_key_id_masked} created at {created_date} it is active and is {key_age} days old." ) expiring_soon = (key_age > config.WARNING_THRESHOLD) and (key_age < config.ALERT_THRESHOLD) expired = key_age > config.ALERT_THRESHOLD in_use = is_key_in_use(iam_client, access_key_id) if expired and not in_use: logger.info( f"{user.get('UserName')}'s access key {access_key_id_masked} has never been used and it's older than {config.ALERT_THRESHOLD} days. Deleting.." ) if not is_dry_run: iam_client.delete_access_key(UserName=user.get("UserName"), AccessKeyId=access_key_id) if not is_dry_run and is_notification_enabled: email_user( ses_client, logger, user.get("Email"), email_subject.render(access_key_deleted=True), email_body_html.render( access_key_deleted=True, full_name=user.get("FullName"), user_name=user.get("UserName"), access_key_id_masked=access_key_id_masked, aws_account_alias=aws_account_alias, threshold=config.ALERT_THRESHOLD, ), email_body_txt.render( access_key_deleted=True, full_name=user.get("FullName"), user_name=user.get("UserName"), access_key_id_masked=access_key_id_masked, aws_account_alias=aws_account_alias, threshold=config.ALERT_THRESHOLD, ), ) elif expired and in_use: logger.info( f"{user.get('UserName')}'s access key {access_key_id_masked} is in use and it's older than {config.ALERT_THRESHOLD} days. Disabling.." ) if not is_dry_run: iam_client.update_access_key( UserName=user.get("UserName"), AccessKeyId=access_key_id, Status="Inactive" ) if not is_dry_run and is_notification_enabled: email_user( ses_client, logger, user.get("Email"), email_subject.render(access_key_expired=True), email_body_html.render( access_key_expired=True, full_name=user.get("FullName"), user_name=user.get("UserName"), access_key_id_masked=access_key_id_masked, aws_account_alias=aws_account_alias, threshold=config.ALERT_THRESHOLD, ), email_body_txt.render( access_key_expired=True, full_name=user.get("FullName"), user_name=user.get("UserName"), access_key_id_masked=access_key_id_masked, aws_account_alias=aws_account_alias, threshold=config.ALERT_THRESHOLD, ), ) elif expiring_soon and in_use: logger.info( f"{user.get('UserName')}'s access key {access_key_id_masked} is in use and it's expiring soon." ) if not is_dry_run and is_notification_enabled: email_user( ses_client, logger, user.get("Email"), email_subject.render(access_key_is_expiring_soon=True), email_body_html.render( access_key_is_expiring_soon=True, full_name=user.get("FullName"), user_name=user.get("UserName"), access_key_id_masked=access_key_id_masked, aws_account_alias=aws_account_alias, threshold=config.ALERT_THRESHOLD, ), email_body_txt.render( access_key_is_expiring_soon=True, full_name=user.get("FullName"), user_name=user.get("UserName"), access_key_id_masked=access_key_id_masked, aws_account_alias=aws_account_alias, threshold=config.ALERT_THRESHOLD, ), ) return {}