diff --git a/prowler/changelog.d/oci-dynamic-group-null-matching-rule.fixed.md b/prowler/changelog.d/oci-dynamic-group-null-matching-rule.fixed.md new file mode 100644 index 0000000000..0bdfef8240 --- /dev/null +++ b/prowler/changelog.d/oci-dynamic-group-null-matching-rule.fixed.md @@ -0,0 +1 @@ +OCI Identity service no longer drops the whole dynamic groups, groups, policies or users listing when the OCI API returns null optional fields such as `matching_rule` diff --git a/prowler/providers/oraclecloud/services/identity/identity_service.py b/prowler/providers/oraclecloud/services/identity/identity_service.py index d36ed5c0ac..5d55fd9577 100644 --- a/prowler/providers/oraclecloud/services/identity/identity_service.py +++ b/prowler/providers/oraclecloud/services/identity/identity_service.py @@ -106,25 +106,24 @@ class Identity(OCIService): identity_client, user.id, compartment.id ) + capabilities = getattr(user, "capabilities", None) + # Check if user can use API keys - can_use_api_keys = ( - user.capabilities.can_use_api_keys - if hasattr(user, "capabilities") - else True + can_use_api_keys = getattr( + capabilities, "can_use_api_keys", None ) + if can_use_api_keys is None: + can_use_api_keys = True # Check if console password is enabled can_use_console_password = ( - user.capabilities.can_use_console_password - if hasattr(user, "capabilities") - else False + getattr(capabilities, "can_use_console_password", None) + or False ) # Check MFA status is_mfa_activated = ( - user.is_mfa_activated - if hasattr(user, "is_mfa_activated") - else False + getattr(user, "is_mfa_activated", None) or False ) self.users.append( @@ -132,19 +131,11 @@ class Identity(OCIService): id=user.id, name=user.name, description=( - user.description or "" - if hasattr(user, "description") - else "" - ), - email=( - user.email or "" - if hasattr(user, "email") - else "" + getattr(user, "description", None) or "" ), + email=(getattr(user, "email", None) or ""), email_verified=( - user.email_verified - if hasattr(user, "email_verified") - else False + getattr(user, "email_verified", None) or False ), compartment_id=compartment.id, time_created=user.time_created, @@ -207,9 +198,7 @@ class Identity(OCIService): auth_tokens.append( AuthToken( id=token.id, - description=( - token.description if hasattr(token, "description") else "" - ), + description=(getattr(token, "description", None) or ""), lifecycle_state=token.lifecycle_state, time_created=token.time_created, time_expires=( @@ -239,9 +228,7 @@ class Identity(OCIService): customer_secret_keys.append( CustomerSecretKey( id=key.id, - display_name=( - key.display_name if hasattr(key, "display_name") else "" - ), + display_name=(getattr(key, "display_name", None) or ""), lifecycle_state=key.lifecycle_state, time_created=key.time_created, time_expires=( @@ -335,9 +322,7 @@ class Identity(OCIService): id=group.id, name=group.name, description=( - group.description - if hasattr(group, "description") - else "" + getattr(group, "description", None) or "" ), compartment_id=compartment.id, time_created=group.time_created, @@ -379,9 +364,7 @@ class Identity(OCIService): id=policy.id, name=policy.name, description=( - policy.description - if hasattr(policy, "description") - else "" + getattr(policy, "description", None) or "" ), compartment_id=compartment.id, statements=policy.statements, @@ -424,15 +407,11 @@ class Identity(OCIService): id=dynamic_group.id, name=dynamic_group.name, description=( - dynamic_group.description or "" - if hasattr(dynamic_group, "description") - else "" + getattr(dynamic_group, "description", None) or "" ), compartment_id=self.audited_tenancy, matching_rule=( - dynamic_group.matching_rule - if hasattr(dynamic_group, "matching_rule") - else "" + getattr(dynamic_group, "matching_rule", None) or "" ), time_created=dynamic_group.time_created, lifecycle_state=dynamic_group.lifecycle_state, diff --git a/tests/providers/oraclecloud/services/identity/identity_service_test.py b/tests/providers/oraclecloud/services/identity/identity_service_test.py index 338a025d52..cb13fd2a83 100644 --- a/tests/providers/oraclecloud/services/identity/identity_service_test.py +++ b/tests/providers/oraclecloud/services/identity/identity_service_test.py @@ -122,6 +122,55 @@ class TestIdentityService: and all(len(d.password_policies) == 1 for d in identity_client.domains) ) + def test_list_dynamic_groups_with_null_optional_fields(self): + """OCI can return `matching_rule` and `description` as null; the + dynamic group must still be retrieved instead of failing the whole + listing with a pydantic ValidationError.""" + with patch( + "prowler.providers.oraclecloud.services.identity.identity_service.Identity.__init__", + return_value=None, + ): + from prowler.providers.oraclecloud.services.identity.identity_service import ( + Identity, + ) + + identity_client = Identity(None) + identity_client.service = "identity" + identity_client.provider = set_mocked_oraclecloud_provider() + identity_client.provider._home_region = "us-ashburn-1" + identity_client.audited_tenancy = "ocid1.tenancy.oc1..aaaaaaaexample" + identity_client.dynamic_groups = [] + identity_client.session_signer = None + identity_client.session_config = None + + regional_client = MagicMock() + regional_client.region = "us-ashburn-1" + + dynamic_group = MagicMock() + dynamic_group.id = "ocid1.dynamicgroup.oc1..aaaaaaaexample" + dynamic_group.name = "prowler-instances" + dynamic_group.description = None + dynamic_group.matching_rule = None + dynamic_group.time_created = datetime.now() + dynamic_group.lifecycle_state = "ACTIVE" + + with ( + patch( + "prowler.providers.oraclecloud.services.identity.identity_service.Identity.__get_client__", + return_value=MagicMock(), + ), + patch( + "prowler.providers.oraclecloud.services.identity.identity_service.oci.pagination.list_call_get_all_results", + return_value=MagicMock(data=[dynamic_group]), + ), + ): + identity_client.__list_dynamic_groups__(regional_client) + + assert len(identity_client.dynamic_groups) == 1 + assert identity_client.dynamic_groups[0].name == "prowler-instances" + assert identity_client.dynamic_groups[0].matching_rule == "" + assert identity_client.dynamic_groups[0].description == "" + def test_list_domains_concurrent_dedupes_and_prefers_home_region(self): """__list_domains__ runs across regions in parallel; the dedupe must stay correct under concurrent calls (no duplicates, home