Add unused access key policy

Author: ebattat

POLICIES.md

@@ -24,6 +24,7 @@ This tool support the following policies:
   monitoring the active connection count.
 * [s3_inactive](cloud_governance/policy/aws/s3_inactive.py): Identify the empty s3 buckets, causing the resource quota
   issues.
+* [unused_access_key](cloud_governance/policy/aws/unused_access_key.py): Identify user with unused active access key, causing the security issue
 * [empty_roles](cloud_governance/policy/aws/empty_roles.py): Identify the empty roles that do not have any attached
   policies to them.
 * [ebs_in_use](cloud_governance/policy/aws/ebs_in_use.py): list in use volumes.

README.md

@@ -35,6 +35,7 @@ List of Policies:
 - zombie_snapshots
 - unused_nat_gateway
 - s3_inactive
+- unused_access_key
 - empty_roles
 - tag_resources
 - tag_iam_user

cloud_governance/common/clouds/aws/iam/iam_operations.py

@@ -5,6 +5,7 @@
 from cloud_governance.common.clouds.aws.utils.common_methods import get_boto3_client
 from cloud_governance.common.clouds.aws.utils.utils import Utils
 from cloud_governance.common.logger.init_logger import logger
+from datetime import datetime, timezone
 
 
 class IAMOperations:
@@ -158,3 +159,141 @@ def untag_role(self, role_name: str, tags: list):
             return True
         except Exception as err:
             raise err
+
+    def tag_user(self, user_name: str, tags: list):
+        """
+        This method tags the IAM user.
+        :param user_name: The name of the IAM user to tag.
+        :param tags: A list of tags to associate with the user.
+        :return: True if tagging is successful, otherwise raises an exception.
+        """
+        try:
+            self.iam_client.tag_user(UserName=user_name, Tags=tags)
+            return True
+        except Exception as err:
+            raise err
+
+    def get_iam_users_access_keys(self):
+        """
+        Retrieves IAM users and summarizes:
+            - Access key status (active/inactive)
+            - Access key age in days
+            - Access key last used in days (or "N/A" if never used)
+            - Tags (as a list of dictionaries)
+            - Most recent key usage: last_activity_days
+            - IAM client region (global context, since IAM is non-regional)
+            - IAM user unique ID: ResourceId
+
+        Returns:
+            dict: {
+                "username": {
+                    "Access key 1": [status, age_days, last_used_days],
+                    "Access key 2": [...],
+                    "last_activity_days": int or "N/A",
+                    "tags": [{"Key": "tag_key", "Value": "tag_value"}, ...],
+                    "region": "us-east-1",
+                    "ResourceId": "AIDAEXAMPLEUSERID"
+                },
+                ...
+            }
+        """
+        result = {}
+        now = datetime.now(timezone.utc)
+        region_name = self.iam_client.meta.region_name or "global"
+
+        paginator = self.iam_client.get_paginator('list_users')
+        for page in paginator.paginate():
+            for user in page['Users']:
+                username = user['UserName']
+                result[username] = {}
+                last_used_days_list = []
+
+                # Access keys
+                access_keys = self.iam_client.list_access_keys(UserName=username)['AccessKeyMetadata']
+                for idx, key in enumerate(access_keys, start=1):
+                    label = f"Access key {idx}"
+                    status = key['Status'].lower()
+                    age_days = (now - key['CreateDate']).days
+
+                    # Get access key last used
+                    try:
+                        response = self.iam_client.get_access_key_last_used(AccessKeyId=key['AccessKeyId'])
+                        last_used_date = response.get('AccessKeyLastUsed', {}).get('LastUsedDate')
+                        if last_used_date:
+                            last_used_days = (now - last_used_date).days
+                            last_used_days_list.append(last_used_days)
+                        else:
+                            last_used_days = "N/A"
+                    except Exception:
+                        last_used_days = "N/A"
+
+                    result[username][label] = [status, age_days, last_used_days]
+
+                # Most recent access key activity
+                result[username]["last_activity_days"] = min(last_used_days_list) if last_used_days_list else "N/A"
+
+                # Tags as list of dicts
+                try:
+                    tag_response = self.iam_client.list_user_tags(UserName=username)
+                    tags = tag_response.get('Tags', [])
+                except Exception:
+                    tags = []
+
+                result[username]["tags"] = tags
+                result[username]["region"] = region_name
+                result[username]["ResourceId"] = user.get('UserId')  # <-- Unique ID
+        return result
+
+    def has_active_access_keys(self, username):
+        """
+        Checks if the given IAM user has any active access keys.
+        When no user access key return False
+        Args:
+            username (str): IAM user name
+
+        Returns:
+            bool: True if any access key is active, False otherwise
+        """
+        try:
+            access_keys = self.iam_client.list_access_keys(UserName=username)['AccessKeyMetadata']
+        except Exception as e:
+            logger.error(f"Failed to list access keys for user '{username}': {e}")
+            return False
+
+        for key in access_keys:
+            if key.get('Status') == 'Active':
+                return True
+        return False
+
+    def deactivate_user_access_keys(self, username):
+        """
+        Deactivates all active access keys for the given IAM user.
+
+        Args:
+            username (str): IAM user name
+        """
+        try:
+            access_keys = self.iam_client.list_access_keys(UserName=username)['AccessKeyMetadata']
+        except Exception as e:
+            logger.error(f"Failed to list access keys for user '{username}': {e}")
+            return
+
+        for idx, key in enumerate(access_keys, start=1):
+            label = f"Access key {idx}"
+            current_status = key['Status'].lower()
+            access_key_id = key['AccessKeyId']
+
+            if current_status == 'active':
+                try:
+                    self.iam_client.update_access_key(
+                        UserName=username,
+                        AccessKeyId=access_key_id,
+                        Status='Inactive'
+                    )
+                    logger.info(f"{label} deactivated for user '{username}'")
+                except Exception as e:
+                    logger.error(f"Failed to deactivate {label} for user '{username}': {e}")
+            else:
+                logger.info(f"{label} is already inactive for user '{username}'")
+
+        logger.info(f"Deactivate access keys for user '{username}' have been processed.")

cloud_governance/common/elasticsearch/modals/policy_es_data.py

@@ -37,6 +37,8 @@ class PolicyEsMetaData(dict):
     launch_time: str = ''
     running_days: int = ''
     create_date: str = ''
+    age_days: int = ''
+    last_activity_days: int = ''
 
     def __post_init__(self):
         """

cloud_governance/common/utils/configs.py

@@ -22,6 +22,8 @@
 EC2_NAMESPACE = 'AWS/EC2'
 CLOUDWATCH_METRICS_AVAILABLE_DAYS = 14
 AWS_DEFAULT_GLOBAL_REGION = 'us-east-1'
+UNUSED_ACCESS_KEY_DAYS = 90
+UNUSED_ACCESS_KEY_MAX_DAY = 1000
 
 # X86 to Graviton
 GRAVITON_MAPPINGS = {

cloud_governance/main/environment_variables.py

@@ -98,7 +98,7 @@ def __init__(self):
                                                                     'ip_unattached', 'unused_nat_gateway',
                                                                     'instance_idle',
                                                                     'ec2_stop', 'ebs_in_use', 'database_idle',
-                                                                    's3_inactive',
+                                                                    's3_inactive', 'unused_access_key',
                                                                     'empty_roles',
                                                                     'zombie_snapshots', 'skipped_resources',
                                                                     'monthly_report', 'optimize_resources_report']

cloud_governance/main/example.yaml

@@ -23,7 +23,7 @@ ATHENA_ACCOUNT_SECRET_KEY: ""
 
 non_cluster_policies: [ 'instance_run', 'unattached_volume', 'cluster_run',
                         'ip_unattached', 'unused_nat_gateway', 'instance_idle',
-                        'ec2_stop', 'ebs_in_use', 'database_idle', 's3_inactive',
+                        'ec2_stop', 'ebs_in_use', 'database_idle', 's3_inactive', 'unused_access_key',
                         'empty_roles', 'tag_resources', 'cost_usage_reports',
                         'zombie_snapshots', 'skipped_resources',
                         'monthly_report', 'optimize_resources_report' ]

cloud_governance/main/main_oerations/main_operations.py

@@ -42,7 +42,7 @@ def run(self):
             # @Todo support for all the aws policies, currently supports ec2_run as urgent requirement
             if self._policy in policies and self._policy in ["instance_run", "unattached_volume", "cluster_run",
                                                              "ip_unattached", "unused_nat_gateway", "instance_idle",
-                                                             "zombie_snapshots", "database_idle", "s3_inactive",
+                                                             "zombie_snapshots", "database_idle", "s3_inactive", "unused_access_key",
                                                              "empty_roles", "tag_resources", "cost_usage_reports"]:
                 source = policy_type
                 if Utils.equal_ignore_case(policy_type, self._public_cloud_name):

cloud_governance/policy/aws/monthly_report.py

@@ -37,7 +37,8 @@ def policy_description(self, policy_name: str):
             'ip_unattached': 'Delete all the elastic_ips that are unused',
             'unused_nat_gateway': ' Delete all unused nat gateways',
             'zombie_snapshots': 'Delete all the snapshots which the AMI does not use',
-            's3_inactive': 'Delete the empty buckets which don’t have any content.',
+            's3_inactive': 'Delete the empty buckets which don’t have any content',
+            'unused_access_key': 'Deactivate user access keys that are still active but have not been used',
             'empty_roles': 'Delete the empty role which does\'t have any policies',
             'zombie_cluster_resource': 'Delete up the cluster resources which are not deleted while cleaning the cluster'
         }

cloud_governance/policy/aws/unused_access_key.py

@@ -0,0 +1,55 @@
+from cloud_governance.common.utils.configs import UNUSED_ACCESS_KEY_DAYS, UNUSED_ACCESS_KEY_MAX_DAY
+from cloud_governance.policy.helpers.aws.aws_policy_operations import AWSPolicyOperations
+
+
+class UnusedAccessKey(AWSPolicyOperations):
+    RESOURCE_ACTION = "DeActivate"
+
+    def __init__(self):
+        super().__init__()
+
+    def run_policy_operations(self):
+        """
+        This method returns a list of users with at least one active access key whose last used date is greater than UNUSED_ACCESS_KEY_DAYS
+        :return:
+        :rtype:
+        """
+        unused_access_keys = []
+        iam_users_access_keys = self._get_iam_users_access_keys()
+        for username, user_data in iam_users_access_keys.items():
+            last_activity_days = user_data['last_activity_days']
+            # Collect age_days only for active access keys
+            age_days_list = [
+                value[1] for key, value in user_data.items()
+                if key.startswith("Access key") and isinstance(value, list) and value[0] == "active"
+            ]
+            # "N/A"/None implies unused access key — fallback to UNUSED_ACCESS_KEY_MAX_DAY
+            if last_activity_days == "N/A":
+                last_activity_days = UNUSED_ACCESS_KEY_MAX_DAY
+            age_days = min(age_days_list) if age_days_list else UNUSED_ACCESS_KEY_MAX_DAY
+            region = user_data['region']
+            user_name = username
+            tags = user_data.get('Tags', [])
+            cleanup_result = False
+            cleanup_days = 0
+            if int(last_activity_days) >= UNUSED_ACCESS_KEY_DAYS and self._has_active_access_keys(user_name) and self.get_skip_policy_value(tags=tags) not in ('NOTDELETE', 'SKIP'):
+                resource_data = self._get_es_schema(resource_id=user_name,
+                                                    user=self.get_tag_name_from_tags(tags=tags, tag_name='User'),
+                                                    skip_policy=self.get_skip_policy_value(tags=tags),
+                                                    cleanup_days=cleanup_days,
+                                                    dry_run=self._dry_run,
+                                                    name=user_name,
+                                                    region=region,
+                                                    cleanup_result=str(cleanup_result),
+                                                    resource_action=self.RESOURCE_ACTION,
+                                                    cloud_name=self._cloud_name,
+                                                    resource_type='UnusedAccessKey',
+                                                    resource_state='Unused',
+                                                    age_days=age_days,
+                                                    last_activity_days=last_activity_days,
+                                                    unit_price=0)
+                unused_access_keys.append(resource_data)
+            if not cleanup_result:
+                self.update_resource_day_count_tag(resource_id=user_name, cleanup_days=cleanup_days, tags=tags)
+
+        return unused_access_keys