diff --git a/cloud_governance/common/clouds/aws/cloudtrail/cloudtrail_operations.py b/cloud_governance/common/clouds/aws/cloudtrail/cloudtrail_operations.py index 6ba67f0d..d6c8dba8 100644 --- a/cloud_governance/common/clouds/aws/cloudtrail/cloudtrail_operations.py +++ b/cloud_governance/common/clouds/aws/cloudtrail/cloudtrail_operations.py @@ -253,6 +253,195 @@ def get_username_by_instance_id_and_time(self, resource_id: str, resource_type: username, event = self.__get_user_by_resource_id(start_time, end_time, resource_id, resource_type, event_type) return self.__check_filter_username(username, event) + def get_username_from_resource_events(self, resource_id: str, iam_users: list, + start_time: datetime = None, end_time: datetime = None, + exclude_users: set = None): + """ + Fallback username lookup: search ALL CloudTrail events for a resource + and return the first username matching a known IAM user. + Handles managed services (ROSA, EKS, etc.) where RunInstances is + performed by a service account but subsequent events (CreateTags, etc.) + may be performed by the actual user via SSO or IAM credentials. + @param resource_id: The AWS resource ID to look up + @param iam_users: List of known IAM usernames to validate against + @param start_time: Optional search window start (defaults to LOOKBACK_DAYS ago) + @param end_time: Optional search window end (defaults to now) + @param exclude_users: Optional set of usernames to skip (automation accounts) + @return: matching IAM username or empty string + """ + if exclude_users is None: + exclude_users = set() + try: + if not start_time: + start_time = datetime.now() - timedelta(days=self.LOOKBACK_DAYS) + if not end_time: + end_time = datetime.now() + responses = self.get_full_responses( + StartTime=start_time, EndTime=end_time, + LookupAttributes=[{ + 'AttributeKey': 'ResourceName', + 'AttributeValue': resource_id + }]) + for event in responses: + username, _ = self.__check_event_is_assumed_role( + event.get('CloudTrailEvent', '')) + if username and username in iam_users and username not in exclude_users: + logger.info(f'Found username {username} from {event.get("EventName")} ' + f'event on resource {resource_id}') + return username + event_username = event.get('Username', '') + if event_username and event_username in iam_users and event_username not in exclude_users: + logger.info(f'Found username {event_username} from {event.get("EventName")} ' + f'event on resource {resource_id}') + return event_username + return '' + except Exception as err: + logger.error(f'Error in get_username_from_resource_events: {err}') + return '' + + def __get_username_from_role_cloudtrail(self, role: dict, iam_users: list): + """ + Look up CloudTrail CreateRole event for a given IAM role and return + the username if it matches a known IAM user. + """ + role_name = role['RoleName'] + role_arn = role['Arn'] + create_date = role.get('CreateDate') + if not create_date: + return '' + end_time = create_date + timedelta(seconds=self.SEARCH_SECONDS) + start_time = create_date - timedelta(seconds=self.SEARCH_SECONDS) + responses = self.__get_cloudtrail_responses( + start_time=start_time, end_time=end_time, + resource_arn=role_arn) + for event in responses: + if event.get('EventName') == 'CreateRole': + username, _ = self.__check_event_is_assumed_role( + event.get('CloudTrailEvent', '')) + if username and username in iam_users: + logger.info(f'Found cluster owner {username} from CreateRole ' + f'event on role {role_name}') + return username + event_username = event.get('Username', '') + if event_username and event_username in iam_users: + logger.info(f'Found cluster owner {event_username} from CreateRole ' + f'event on role {role_name}') + return event_username + return '' + + def __get_username_from_create_role_events(self, start_time: datetime, + end_time: datetime, + iam_users: list): + """ + Search CloudTrail for CreateRole events of OpenShift operator roles + within a time window. Handles the case where roles were deleted from + IAM but creation events still exist in CloudTrail (90-day retention). + """ + try: + events = [] + kwargs = { + 'LookupAttributes': [{ + 'AttributeKey': 'EventName', + 'AttributeValue': 'CreateRole' + }], + 'StartTime': start_time, 'EndTime': end_time, + 'MaxResults': 50 + } + while True: + response = self.__global_cloudtrail.lookup_events(**kwargs) + events.extend(response.get('Events', [])) + next_token = response.get('NextToken') + if not next_token: + break + kwargs['NextToken'] = next_token + for event in events: + resources = event.get('Resources', []) + role_names = [r.get('ResourceName', '') for r in resources + if r.get('ResourceType') == 'AWS::IAM::Role'] + if not any('openshift' in n.lower() for n in role_names): + continue + username, _ = self.__check_event_is_assumed_role( + event.get('CloudTrailEvent', '')) + if username and username in iam_users: + logger.info(f'Found cluster owner {username} from ' + f'CloudTrail CreateRole event (deleted role)') + return username + event_username = event.get('Username', '') + if event_username and event_username in iam_users: + logger.info(f'Found cluster owner {event_username} from ' + f'CloudTrail CreateRole event (deleted role)') + return event_username + except Exception as err: + logger.error(f'Error searching CreateRole events: {err}') + return '' + + def get_username_from_cluster_role(self, cluster_id: str, iam_users: list, + launch_time: datetime = None): + """ + Trace a cluster's ownership through its IAM roles. + Uses three strategies: + 1. Direct match: find roles whose names contain the cluster ID + (works for IPI/UPI clusters) + 2. Temporal match: find ROSA/OpenShift operator roles created + shortly before the cluster's instance launch time + (works for ROSA STS where roles use cluster name, not infra ID) + 3. CloudTrail fallback: search CreateRole events directly for + OpenShift roles in the time window (handles deleted roles) + Works for both SSO (AssumedRole) and IAM user identities. + @param cluster_id: The cluster infra ID (e.g. 'u0s3a7y7o5y9c6w-x56mc') + @param iam_users: List of known IAM usernames to validate against + @param launch_time: Instance launch time for temporal role matching + @return: matching IAM username or empty string + """ + ROSA_ROLE_WINDOW_SECONDS = 21600 # 6 hours before launch + try: + paginator = self.__iam_client.get_paginator('list_roles') + roles = [] + for page in paginator.paginate(PaginationConfig={'PageSize': 1000}): + roles.extend(page.get('Roles', [])) + cluster_roles = [r for r in roles if cluster_id in r.get('RoleName', '')] + if cluster_roles: + for role in cluster_roles: + username = self.__get_username_from_role_cloudtrail(role, iam_users) + if username: + return username + + if not launch_time: + return '' + candidate_roles = [] + for role in roles: + role_name = role.get('RoleName', '') + if 'openshift' not in role_name.lower(): + continue + create_date = role.get('CreateDate') + if not create_date: + continue + launch_naive = launch_time.replace(tzinfo=None) + create_naive = create_date.replace(tzinfo=None) + diff_seconds = (launch_naive - create_naive).total_seconds() + if 0 <= diff_seconds <= ROSA_ROLE_WINDOW_SECONDS: + candidate_roles.append(role) + if candidate_roles: + candidate_roles.sort(key=lambda r: r.get('CreateDate', ''), reverse=True) + resolved_users = set() + for role in candidate_roles[:6]: + username = self.__get_username_from_role_cloudtrail(role, iam_users) + if username: + resolved_users.add(username) + if len(resolved_users) == 1: + return resolved_users.pop() + + ct_start = launch_time - timedelta(seconds=ROSA_ROLE_WINDOW_SECONDS) + username = self.__get_username_from_create_role_events( + start_time=ct_start, end_time=launch_time, + iam_users=iam_users) + if username: + return username + return '' + except Exception as err: + logger.error(f'Error in get_username_from_cluster_role: {err}') + return '' + def get_stop_time(self, resource_id: str, event_name: str): """ This method return the time of when instance is stopped diff --git a/cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_operations.py b/cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_operations.py index a4f43b4c..4a831f62 100644 --- a/cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_operations.py +++ b/cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_operations.py @@ -6,6 +6,7 @@ from cloud_governance.common.clouds.aws.ec2.ec2_operations import EC2Operations from cloud_governance.common.clouds.aws.iam.iam_operations import IAMOperations from cloud_governance.common.clouds.aws.utils.utils import Utils +from cloud_governance.common.logger.init_logger import logger class TagClusterOperations: @@ -29,9 +30,25 @@ def __init__(self, region: str, input_tags: dict = None, cluster_name: str = No self.cluster_name = cluster_name self.input_tags = input_tags self.cloudtrail = CloudTrailOperations(region_name='us-east-1') - self._get_username_from_instance_id_and_time = CloudTrailOperations(region_name=region).get_username_by_instance_id_and_time + self._regional_cloudtrail = CloudTrailOperations(region_name=region) + self._get_username_from_instance_id_and_time = self._regional_cloudtrail.get_username_by_instance_id_and_time self.dry_run = dry_run self.iam_users = self.iam_operations.get_iam_users_list() + self._automation_user = self._get_automation_username() + + @staticmethod + def _get_automation_username(): + """ + Identify the current caller (the automation/service account running + this policy) so it can be excluded from CloudTrail fallback searches. + """ + try: + sts = boto3.client('sts') + identity = sts.get_caller_identity() + arn = identity.get('Arn', '') + return arn.split('/')[-1] if '/' in arn else '' + except Exception: + return '' def _input_tags_list_builder(self): """ @@ -89,12 +106,61 @@ def get_user_name_from_name_tag(self, tags: list): return user return None - def get_username(self, start_time: datetime, resource_id: str, resource_type: str, tags: list): + def get_username(self, start_time: datetime, resource_id: str, resource_type: str, + tags: list, cluster_id: str = ''): """ - This method returns the username + This method returns the username using multiple strategies: + 1. Check existing tags (User tag, Name tag) + 2. CloudTrail - resource creation event (RunInstances, etc.) + 3. CloudTrail - any event on this resource by a known IAM user + (handles managed services like ROSA where service accounts create + resources but user events like CreateTags may exist) + 4. CloudTrail - trace cluster IAM roles to find who created them + (handles ROSA/OSD where instances are created by service accounts + but IAM roles are created by the user) :return: """ + exclude = {self._automation_user} if self._automation_user else set() iam_username = self.get_user_name_from_name_tag(tags=tags) if not iam_username: - return self._get_username_from_instance_id_and_time(start_time=start_time, resource_id=resource_id, resource_type=resource_type) + ct_username = self._get_username_from_instance_id_and_time( + start_time=start_time, resource_id=resource_id, + resource_type=resource_type) + if ct_username == 'AutoScaling': + return ct_username + if ct_username and ct_username in self.iam_users and ct_username not in exclude: + return ct_username + if cluster_id: + iam_username = self.cloudtrail.get_username_from_cluster_role( + cluster_id=cluster_id, iam_users=self.iam_users, + launch_time=start_time) + if iam_username: + logger.info(f'Resolved cluster owner {iam_username} via IAM ' + f'role lookup for cluster {cluster_id}') + return iam_username + iam_username = self._regional_cloudtrail.get_username_from_resource_events( + resource_id=resource_id, iam_users=self.iam_users, + start_time=start_time, exclude_users=exclude) return iam_username + + def get_username_from_cluster_instances(self, resources: list, cluster_name: str): + """ + Search all instances in a cluster for a valid User tag. + If any instance already has a tagged owner, return that username. + Handles managed services (ROSA, EKS) where CloudTrail cannot attribute + the creator - if at least one instance was tagged, propagate to siblings. + @param resources: list of instance groups from the cluster + @param cluster_name: the cluster identifier to match + @return: username string or None + """ + for instance_group in resources: + for item in instance_group: + tags = item.get('Tags', []) + for tag in tags: + if any(prefix in tag.get('Key', '') for prefix in self.cluster_prefix): + if cluster_name in tag.get('Key', ''): + user = self.ec2_operations.get_tag_value_from_tags( + tags=tags, tag_name='User') + if user and user != 'NA' and user in self.iam_users: + return user + return None diff --git a/cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_resouces.py b/cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_resouces.py index 9356c9cf..b08b5f27 100644 --- a/cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_resouces.py +++ b/cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_resouces.py @@ -312,7 +312,11 @@ def update_cluster_tags(self, resources: list): else: username = self.get_username(start_time=item.get('LaunchTime'), resource_id=instance_id, - resource_type='AWS::EC2::Instance', tags=tags) + resource_type='AWS::EC2::Instance', tags=tags, + cluster_id=cluster_name) + if not username: + username = self.get_username_from_cluster_instances( + resources=resources, cluster_name=cluster_name) if username: if username == 'AutoScaling': add_tags.extend(self._fill_na_tags(user=username)) diff --git a/cloud_governance/policy/policy_operations/aws/tag_non_cluster/non_cluster_operations.py b/cloud_governance/policy/policy_operations/aws/tag_non_cluster/non_cluster_operations.py index f7b3519d..10cf3b88 100644 --- a/cloud_governance/policy/policy_operations/aws/tag_non_cluster/non_cluster_operations.py +++ b/cloud_governance/policy/policy_operations/aws/tag_non_cluster/non_cluster_operations.py @@ -24,6 +24,17 @@ def __init__(self, region: str = 'us-east-2', dry_run: str = 'yes', input_tags: self.ec2_operations = EC2Operations(region=region) self.utils = Utils(region=region) self.iam_users = self.iam_client.get_iam_users_list() + self._automation_user = self._get_automation_username() + + @staticmethod + def _get_automation_username(): + try: + sts = boto3.client('sts') + identity = sts.get_caller_identity() + arn = identity.get('Arn', '') + return arn.split('/')[-1] if '/' in arn else '' + except Exception: + return '' def _get_instances_data(self, instance_id: str = ''): """ @@ -209,14 +220,27 @@ def get_user_name_from_name_tag(self, tags: list = None, resource_name: str = No def get_username(self, start_time: datetime, resource_id: str, resource_type: str, tags: list, resource_name: str = '', end_time: datetime = None): """ - This method returns the username + This method returns the username using multiple strategies: + 1. Check existing tags (Name tag) + 2. CloudTrail - resource creation event + 3. CloudTrail - any event on this resource by a known IAM user + (handles managed services where service accounts create resources) :return: """ iam_username = self.get_user_name_from_name_tag(tags=tags, resource_name=resource_name) if not iam_username: iam_username = self.get_user_name_from_name_tag(resource_name=resource_name) if not iam_username: - return self._get_username_from_cloudtrail(start_time=start_time, resource_id=resource_id, resource_type=resource_type, end_time=end_time) + ct_username = self._get_username_from_cloudtrail( + start_time=start_time, resource_id=resource_id, + resource_type=resource_type, end_time=end_time) + exclude = {self._automation_user} if self._automation_user else set() + if ct_username and ct_username in self.iam_users and ct_username not in exclude: + return ct_username + iam_username = self.cloudtrail.get_username_from_resource_events( + resource_id=resource_id, iam_users=self.iam_users, + start_time=start_time, end_time=end_time, + exclude_users=exclude) return iam_username def validate_existing_tag(self, tags: list): @@ -225,7 +249,9 @@ def validate_existing_tag(self, tags: list): @param tags: @return: """ - check_tags = ['User', 'Project', 'Manager', 'Owner', 'Email'] + check_tags = ['User', 'Project', 'Manager', 'Owner', 'Email', 'LaunchTime'] + if self.input_tags: + check_tags.extend(key for key in self.input_tags if key not in check_tags) tag_count = 0 if tags: for tag in tags: diff --git a/cloud_governance/policy/policy_operations/aws/tag_non_cluster/remove_non_cluster_tags.py b/cloud_governance/policy/policy_operations/aws/tag_non_cluster/remove_non_cluster_tags.py index 388b64ca..62f2e478 100644 --- a/cloud_governance/policy/policy_operations/aws/tag_non_cluster/remove_non_cluster_tags.py +++ b/cloud_governance/policy/policy_operations/aws/tag_non_cluster/remove_non_cluster_tags.py @@ -17,7 +17,11 @@ def __get_instance_tags(self, launch_time: datetime, instance_id: str, tags: lis @param tags: @return: """ - username = self._get_username_from_cloudtrail(start_time=launch_time, resource_id=instance_id, resource_type='AWS::EC2::Instance') + username = self.ec2_operations.get_tag_value_from_tags(tags=tags or [], tag_name='User') + if not username or username == self.NA_VALUE: + username = self.get_username( + start_time=launch_time, resource_id=instance_id, + resource_type='AWS::EC2::Instance', tags=tags or []) search_tags = [] user_tags = self.iam_client.get_user_tags(username=username) if not username: diff --git a/tests/unittest/cloud_governance/aws/tag_cluster/test_tag_cluster_operations.py b/tests/unittest/cloud_governance/aws/tag_cluster/test_tag_cluster_operations.py new file mode 100644 index 00000000..32ff3119 --- /dev/null +++ b/tests/unittest/cloud_governance/aws/tag_cluster/test_tag_cluster_operations.py @@ -0,0 +1,90 @@ +from datetime import datetime, timezone +from unittest.mock import Mock + +from cloud_governance.policy.policy_operations.aws.tag_cluster.tag_cluster_operations import TagClusterOperations + +CLUSTER_OWNER = 'cluster-owner' +LEGACY_TAGGER = 'legacy-tagger' +AUTOMATION_USER = 'automation-user' + + +def _make_tag_cluster_ops(**overrides): + """Build TagClusterOperations without calling __init__ (avoids real AWS calls).""" + ops = TagClusterOperations.__new__(TagClusterOperations) + ops.iam_users = overrides.get('iam_users', [CLUSTER_OWNER, LEGACY_TAGGER]) + ops._automation_user = overrides.get('_automation_user', AUTOMATION_USER) + ops.get_user_name_from_name_tag = Mock(return_value=overrides.get('name_tag_user')) + ops._get_username_from_instance_id_and_time = Mock( + return_value=overrides.get('instance_ct_user')) + ops.cloudtrail = Mock() + ops.cloudtrail.get_username_from_cluster_role = Mock( + return_value=overrides.get('cluster_role_user', '')) + ops._regional_cloudtrail = Mock() + ops._regional_cloudtrail.get_username_from_resource_events = Mock( + return_value=overrides.get('resource_events_user', '')) + return ops + + +class TestTagClusterOperationsGetUsername: + LAUNCH_TIME = datetime(2026, 6, 30, 14, 6, 46, tzinfo=timezone.utc) + RESOURCE_ID = 'i-0123456789abcdef0' + CLUSTER_ID = 'test-infra-id-x56mc' + + def test_returns_autoscaling_without_iam_validation(self): + ops = _make_tag_cluster_ops(instance_ct_user='AutoScaling') + result = ops.get_username( + start_time=self.LAUNCH_TIME, resource_id=self.RESOURCE_ID, + resource_type='AWS::EC2::Instance', tags=[], cluster_id=self.CLUSTER_ID) + assert result == 'AutoScaling' + + def test_skips_automation_user_from_instance_cloudtrail(self): + ops = _make_tag_cluster_ops( + iam_users=[CLUSTER_OWNER, LEGACY_TAGGER, AUTOMATION_USER], + instance_ct_user=AUTOMATION_USER, + cluster_role_user=CLUSTER_OWNER) + result = ops.get_username( + start_time=self.LAUNCH_TIME, resource_id=self.RESOURCE_ID, + resource_type='AWS::EC2::Instance', tags=[], cluster_id=self.CLUSTER_ID) + assert result == CLUSTER_OWNER + ops.cloudtrail.get_username_from_cluster_role.assert_called_once() + + def test_cluster_role_resolved_before_resource_events_fallback(self): + ops = _make_tag_cluster_ops( + instance_ct_user='178000000000', + cluster_role_user=CLUSTER_OWNER, + resource_events_user=LEGACY_TAGGER) + result = ops.get_username( + start_time=self.LAUNCH_TIME, resource_id=self.RESOURCE_ID, + resource_type='AWS::EC2::Instance', tags=[], cluster_id=self.CLUSTER_ID) + assert result == CLUSTER_OWNER + ops._regional_cloudtrail.get_username_from_resource_events.assert_not_called() + + def test_resource_events_fallback_excludes_automation_user(self): + ops = _make_tag_cluster_ops( + instance_ct_user='178000000000', + cluster_role_user='', + resource_events_user=CLUSTER_OWNER) + result = ops.get_username( + start_time=self.LAUNCH_TIME, resource_id=self.RESOURCE_ID, + resource_type='AWS::EC2::Instance', tags=[], cluster_id=self.CLUSTER_ID) + assert result == CLUSTER_OWNER + call_kwargs = ops._regional_cloudtrail.get_username_from_resource_events.call_args.kwargs + assert call_kwargs['exclude_users'] == {AUTOMATION_USER} + + +class TestTagClusterOperationsClusterInstanceFallback: + def test_get_username_from_cluster_instances_returns_tagged_peer(self): + ops = TagClusterOperations.__new__(TagClusterOperations) + ops.cluster_prefix = ['kubernetes.io/cluster/'] + ops.iam_users = [CLUSTER_OWNER] + ops.ec2_operations = Mock() + ops.ec2_operations.get_tag_value_from_tags = Mock(return_value=CLUSTER_OWNER) + resources = [[{ + 'Tags': [ + {'Key': 'kubernetes.io/cluster/test-infra-id-x56mc', 'Value': 'owned'}, + {'Key': 'User', 'Value': CLUSTER_OWNER}, + ] + }]] + result = ops.get_username_from_cluster_instances( + resources=resources, cluster_name='test-infra-id-x56mc') + assert result == CLUSTER_OWNER diff --git a/tests/unittest/cloud_governance/aws/tag_non_cluster/test_non_cluster_operations.py b/tests/unittest/cloud_governance/aws/tag_non_cluster/test_non_cluster_operations.py new file mode 100644 index 00000000..950f0871 --- /dev/null +++ b/tests/unittest/cloud_governance/aws/tag_non_cluster/test_non_cluster_operations.py @@ -0,0 +1,82 @@ +from datetime import datetime, timezone +from unittest.mock import Mock + +from cloud_governance.policy.policy_operations.aws.tag_non_cluster.non_cluster_operations import NonClusterOperations + +CLUSTER_OWNER = 'cluster-owner' +LEGACY_TAGGER = 'legacy-tagger' +AUTOMATION_USER = 'automation-user' + + +def _make_non_cluster_ops(**overrides): + ops = NonClusterOperations.__new__(NonClusterOperations) + ops.iam_users = overrides.get('iam_users', [CLUSTER_OWNER, LEGACY_TAGGER]) + ops._automation_user = overrides.get('_automation_user', AUTOMATION_USER) + ops.get_user_name_from_name_tag = Mock(return_value=overrides.get('name_tag_user')) + ops._get_username_from_cloudtrail = Mock(return_value=overrides.get('cloudtrail_user')) + ops.cloudtrail = Mock() + ops.cloudtrail.get_username_from_resource_events = Mock( + return_value=overrides.get('resource_events_user', '')) + return ops + + +class TestNonClusterOperationsGetUsername: + START_TIME = datetime(2026, 6, 30, 14, 0, 0, tzinfo=timezone.utc) + + def test_excludes_automation_user_from_primary_cloudtrail(self): + ops = _make_non_cluster_ops( + iam_users=[CLUSTER_OWNER, LEGACY_TAGGER, AUTOMATION_USER], + cloudtrail_user=AUTOMATION_USER) + result = ops.get_username( + start_time=self.START_TIME, resource_id='vol-abc123', + resource_type='AWS::EC2::Volume', tags=[]) + assert result == '' + ops.cloudtrail.get_username_from_resource_events.assert_called_once() + + def test_returns_valid_user_from_primary_cloudtrail(self): + ops = _make_non_cluster_ops(cloudtrail_user=CLUSTER_OWNER) + result = ops.get_username( + start_time=self.START_TIME, resource_id='vol-abc123', + resource_type='AWS::EC2::Volume', tags=[]) + assert result == CLUSTER_OWNER + ops.cloudtrail.get_username_from_resource_events.assert_not_called() + + def test_resource_events_fallback_excludes_automation_user(self): + ops = _make_non_cluster_ops( + cloudtrail_user='', + resource_events_user=CLUSTER_OWNER) + result = ops.get_username( + start_time=self.START_TIME, resource_id='vol-abc123', + resource_type='AWS::EC2::Volume', tags=[]) + assert result == CLUSTER_OWNER + call_kwargs = ops.cloudtrail.get_username_from_resource_events.call_args.kwargs + assert call_kwargs['exclude_users'] == {AUTOMATION_USER} + + +class TestNonClusterOperationsValidateExistingTag: + def test_returns_false_when_budget_tag_missing(self): + ops = NonClusterOperations.__new__(NonClusterOperations) + ops.input_tags = {'Budget': 'PERF-DEPT'} + tags = [ + {'Key': 'User', 'Value': 'cluster-owner'}, + {'Key': 'Project', 'Value': 'test'}, + {'Key': 'Manager', 'Value': 'manager'}, + {'Key': 'Owner', 'Value': 'owner'}, + {'Key': 'Email', 'Value': 'cluster-owner@redhat.com'}, + {'Key': 'LaunchTime', 'Value': '2026/06/30 14:00:00'}, + ] + assert ops.validate_existing_tag(tags=tags) is False + + def test_returns_true_when_all_required_tags_present(self): + ops = NonClusterOperations.__new__(NonClusterOperations) + ops.input_tags = {'Budget': 'PERF-DEPT'} + tags = [ + {'Key': 'User', 'Value': 'cluster-owner'}, + {'Key': 'Project', 'Value': 'test'}, + {'Key': 'Manager', 'Value': 'manager'}, + {'Key': 'Owner', 'Value': 'owner'}, + {'Key': 'Email', 'Value': 'cluster-owner@redhat.com'}, + {'Key': 'LaunchTime', 'Value': '2026/06/30 14:00:00'}, + {'Key': 'Budget', 'Value': 'PERF-DEPT'}, + ] + assert ops.validate_existing_tag(tags=tags) is True diff --git a/tests/unittest/cloud_governance/common/clouds/aws/cloudtrail/test_cloudtrail_operations.py b/tests/unittest/cloud_governance/common/clouds/aws/cloudtrail/test_cloudtrail_operations.py index 9a0179a6..9c2180a0 100644 --- a/tests/unittest/cloud_governance/common/clouds/aws/cloudtrail/test_cloudtrail_operations.py +++ b/tests/unittest/cloud_governance/common/clouds/aws/cloudtrail/test_cloudtrail_operations.py @@ -1,4 +1,5 @@ import json +from datetime import datetime, timezone from unittest.mock import Mock, patch import pytest @@ -349,3 +350,128 @@ def test_arn_parsing_different_session_formats(self, mock_boto3): for arn, expected_username in test_cases: session_name = arn.split('/')[-1] assert session_name == expected_username, f"Failed to parse {arn}" + + +class TestCloudTrailOperationsRosaTagging: + """Tests for ROSA cluster ownership resolution added in PR #1007.""" + + CLUSTER_OWNER = 'cluster-owner' + LEGACY_TAGGER = 'legacy-tagger' + LAUNCH_TIME = datetime(2026, 6, 30, 14, 6, 46, tzinfo=timezone.utc) + IAM_USERS = [CLUSTER_OWNER, LEGACY_TAGGER] + + @patch('cloud_governance.common.clouds.aws.cloudtrail.cloudtrail_operations.boto3') + def test_get_username_from_resource_events_skips_excluded_users(self, mock_boto3): + cloudtrail_ops = CloudTrailOperations(region_name='us-east-1') + cloudtrail_ops.get_full_responses = Mock(return_value=[ + { + 'EventName': 'CreateTags', + 'Username': self.LEGACY_TAGGER, + 'CloudTrailEvent': '{}', + }, + { + 'EventName': 'CreateTags', + 'Username': self.CLUSTER_OWNER, + 'CloudTrailEvent': '{}', + }, + ]) + result = cloudtrail_ops.get_username_from_resource_events( + resource_id='i-0123456789abcdef0', + iam_users=self.IAM_USERS, + start_time=self.LAUNCH_TIME, + exclude_users={self.LEGACY_TAGGER}) + assert result == self.CLUSTER_OWNER + + @patch('cloud_governance.common.clouds.aws.cloudtrail.cloudtrail_operations.boto3') + def test_get_username_from_cluster_role_direct_match(self, mock_boto3): + mock_iam = Mock() + mock_paginator = Mock() + mock_paginator.paginate.return_value = [{ + 'Roles': [{ + 'RoleName': 'test-infra-id-x56mc-master-role', + 'Arn': 'arn:aws:iam::123456789012:role/test-infra-id-x56mc-master-role', + 'CreateDate': self.LAUNCH_TIME, + }] + }] + mock_iam.get_paginator.return_value = mock_paginator + mock_boto3.client.side_effect = lambda service, **kwargs: { + 'cloudtrail': Mock(), + 'iam': mock_iam, + }[service] + + cloudtrail_ops = CloudTrailOperations(region_name='us-east-1') + cloudtrail_ops._CloudTrailOperations__get_username_from_role_cloudtrail = Mock( + return_value=self.CLUSTER_OWNER) + + result = cloudtrail_ops.get_username_from_cluster_role( + cluster_id='test-infra-id-x56mc', + iam_users=self.IAM_USERS, + launch_time=self.LAUNCH_TIME) + assert result == self.CLUSTER_OWNER + + @patch('cloud_governance.common.clouds.aws.cloudtrail.cloudtrail_operations.boto3') + def test_get_username_from_cluster_role_ambiguous_temporal_match_returns_empty( + self, mock_boto3): + role_create = datetime(2026, 6, 30, 13, 0, 0, tzinfo=timezone.utc) + mock_iam = Mock() + mock_paginator = Mock() + mock_paginator.paginate.return_value = [{ + 'Roles': [ + { + 'RoleName': 'cluster-a-openshift-machine-api', + 'Arn': 'arn:aws:iam::123456789012:role/cluster-a-openshift-machine-api', + 'CreateDate': role_create, + }, + { + 'RoleName': 'cluster-b-openshift-machine-api', + 'Arn': 'arn:aws:iam::123456789012:role/cluster-b-openshift-machine-api', + 'CreateDate': role_create, + }, + ] + }] + mock_iam.get_paginator.return_value = mock_paginator + mock_boto3.client.side_effect = lambda service, **kwargs: { + 'cloudtrail': Mock(), + 'iam': mock_iam, + }[service] + + cloudtrail_ops = CloudTrailOperations(region_name='us-east-1') + cloudtrail_ops._CloudTrailOperations__get_username_from_role_cloudtrail = Mock( + side_effect=[self.CLUSTER_OWNER, 'otheruser']) + cloudtrail_ops._CloudTrailOperations__get_username_from_create_role_events = Mock( + return_value='') + + result = cloudtrail_ops.get_username_from_cluster_role( + cluster_id='unknown-infra-id', + iam_users=self.IAM_USERS + ['otheruser'], + launch_time=self.LAUNCH_TIME) + assert result == '' + cloudtrail_ops._CloudTrailOperations__get_username_from_create_role_events.assert_called_once() + + @patch('cloud_governance.common.clouds.aws.cloudtrail.cloudtrail_operations.boto3') + def test_get_username_from_create_role_events_paginates_global_cloudtrail( + self, mock_boto3): + mock_global_ct = Mock() + mock_global_ct.lookup_events.side_effect = [ + {'Events': [], 'NextToken': 'page2'}, + {'Events': [{ + 'EventName': 'CreateRole', + 'Username': self.CLUSTER_OWNER, + 'Resources': [{ + 'ResourceType': 'AWS::IAM::Role', + 'ResourceName': 'rosa-test-openshift-machine-api', + }], + 'CloudTrailEvent': '{}', + }]}, + ] + mock_boto3.client.return_value = Mock() + + cloudtrail_ops = CloudTrailOperations(region_name='us-west-2') + cloudtrail_ops._CloudTrailOperations__global_cloudtrail = mock_global_ct + + start = datetime(2026, 6, 30, 8, 0, 0, tzinfo=timezone.utc) + end = datetime(2026, 6, 30, 14, 0, 0, tzinfo=timezone.utc) + result = cloudtrail_ops._CloudTrailOperations__get_username_from_create_role_events( + start_time=start, end_time=end, iam_users=self.IAM_USERS) + assert result == self.CLUSTER_OWNER + assert mock_global_ct.lookup_events.call_count == 2