fix(oci): fix identity clients (#10524)

Co-authored-by: rchotacode <32524742+rchotacode@users.noreply.github.com>
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

api/pyproject.toml

@@ -25,7 +25,7 @@ dependencies = [
   "defusedxml==0.7.1",
   "gunicorn==23.0.0",
   "lxml==5.3.2",
-  "prowler @ git+https://github.com/prowler-cloud/prowler.git@master",
+  "prowler @ git+https://github.com/prowler-cloud/prowler.git@v5.22",
   "psycopg2-binary==2.9.9",
   "pytest-celery[redis] (>=1.0.1,<2.0.0)",
   "sentry-sdk[django] (>=2.20.0,<3.0.0)",
@@ -50,7 +50,7 @@ name = "prowler-api"
 package-mode = false
 # Needed for the SDK compatibility
 requires-python = ">=3.11,<3.13"
-version = "1.23.0"
+version = "1.23.1"
 
 [project.scripts]
 celery = "src.backend.config.settings.celery"

api/src/backend/api/specs/v1.yaml

@@ -1,7 +1,7 @@
 openapi: 3.0.3
 info:
   title: Prowler API
-  version: 1.23.0
+  version: 1.23.1
   description: |-
     Prowler API specification.
 

api/src/backend/api/v1/views.py

@@ -409,7 +409,7 @@ class SchemaView(SpectacularAPIView):
 
     def get(self, request, *args, **kwargs):
         spectacular_settings.TITLE = "Prowler API"
-        spectacular_settings.VERSION = "1.23.0"
+        spectacular_settings.VERSION = "1.23.1"
         spectacular_settings.DESCRIPTION = (
             "Prowler API specification.\n\nThis file is auto-generated."
         )

docs/getting-started/installation/prowler-app.mdx

@@ -121,8 +121,8 @@ To update the environment file:
 Edit the `.env` file and change version values:
 
 ```env
-PROWLER_UI_VERSION="5.21.0"
-PROWLER_API_VERSION="5.21.0"
+PROWLER_UI_VERSION="5.22.0"
+PROWLER_API_VERSION="5.22.0"
 ```
 
 <Note>

prowler/CHANGELOG.md

@@ -2,6 +2,21 @@
 
 All notable changes to the **Prowler SDK** are documented in this file.
 
+## [5.22.1] (Prowler UNRELEASED)
+
+### 🐞 Fixed
+
+- AWS global services (CloudFront, Route53, Shield, FMS) now use the partition's global region instead of the profile's default region [(#10458)](https://github.com/prowler-cloud/prowler/issues/10458)
+- Oracle Cloud `events_rule_idp_group_mapping_changes` now recognizes the CIS 3.1 `add/remove` event names to avoid false positives [(#10416)](https://github.com/prowler-cloud/prowler/pull/10416)
+- Oracle Cloud password policy checks now exclude immutable system-managed policies (`SimplePasswordPolicy`, `StandardPasswordPolicy`) to avoid false positives [(#10453)](https://github.com/prowler-cloud/prowler/pull/10453)
+- Oracle Cloud `kms_key_rotation_enabled` now checks current key version age to avoid false positives on vaults without auto-rotation support [(#10450)](https://github.com/prowler-cloud/prowler/pull/10450)
+- Oracle Cloud patch for filestorage, blockstorage, kms, and compute services in OCI to allow for region scanning outside home [(#10455)](https://github.com/prowler-cloud/prowler/pull/10472)
+- Oracle cloud provider now supports multi-region filtering [(#10435)](https://github.com/prowler-cloud/prowler/pull/10473)
+- `prowler image --registry` failing with `ImageNoImagesProvidedError` due to registry arguments not being forwarded to `ImageProvider` in `init_global_provider` [(#10457)](https://github.com/prowler-cloud/prowler/issues/10457)
+- Oracle Cloud multi-region support for identity client configuration in blockstorage, identity, and filestorage services [(#10519)](https://github.com/prowler-cloud/prowler/pull/10520)
+
+---
+
 ## [5.22.0] (Prowler v5.22.0)
 
 ### 🐞 Fixed

prowler/compliance/oraclecloud/cis_3.1_oraclecloud.json

@@ -662,8 +662,8 @@
           "Description": "It is recommended to setup an Event Rule and Notification that gets triggered when Identity Provider Group Mappings are created, updated or deleted. Event Rules are compartment scoped and will detect events in child compartments. It is recommended to create the Event rule at the root compartment level.",
           "RationaleStatement": "IAM Policies govern access to all resources within an OCI Tenancy. IAM Policies use OCI Groups for assigning the privileges. Identity Provider Groups could be mapped to OCI Groups to assign privileges to federated users in OCI. Monitoring and alerting on changes to Identity Provider Group mappings will help in identifying changes to the security posture.",
           "ImpactStatement": "There is no performance impact when enabling the above described features but depending on the amount of notifications sent per month there may be a cost associated.",
-          "RemediationProcedure": "**From Console:**1. Go to the `Events Service` page: [https://cloud.oracle.com/events/rules](https://cloud.oracle.com/events/rules)2. Select the `compartment` that should host the rule3. Click `Create Rule`4. Provide a `Display Name` and `Description`5. Create a Rule Condition by selecting `Identity` in the Service Name Drop-down and selecting `Idp Group Mapping – Create`, `Idp Group Mapping – Delete` and `Idp Group Mapping – Update`6. In the `Actions` section select `Notifications` as Action Type7. Select the `Compartment` that hosts the Topic to be used.8. Select the `Topic` to be used9. Optionally add Tags to the Rule10. Click `Create Rule`**From CLI:**1. Find the `topic-id` of the topic the Event Rule should use for sending notifications by using the topic `name` and `Compartment OCID````oci ons topic list --compartment-id <compartment-ocid> --all --query data [?name=='<topic-name>'].{name:name,topic_id:\\topic-id\\} --output table```2. Create a JSON file to be used when creating the Event Rule. Replace topic id, display name, description and compartment OCID.```{ actions: { actions: [ { actionType: ONS, isEnabled: true, topicId: <topic-id> }] }, condition:{\\eventType\\:[\\com.oraclecloud.identitycontrolplane.addidpgroupmapping\\,\\com.oraclecloud.identitycontrolplane.removeidpgroupmapping\\,\\com.oraclecloud.identitycontrolplane.updateidpgroupmapping\\],\\data\\:{}}, displayName: <display-name>, description: <description>, isEnabled: true, compartmentId: <compartment-ocid>}```3. Create the actual event rule```oci events rule create --from-json file://event_rule.json```4. Note in the JSON returned that it lists the parameters specified in the JSON file provided and that there is an OCID provided for the Event Rule",
-          "AuditProcedure": "**From Console:**1. Go to the Events Service page: [https://cloud.oracle.com/events/rules](https://cloud.oracle.com/events/rules)2. Select the `Compartment` that hosts the rules3. Find and click the `Rule` that handles `Idp Group Mapping` Changes (if any)4. Ensure the `Rule` is `ACTIVE`5. Click the `Edit Rule` button and verify that the `RuleConditions` section contains a condition for the Service `Identity` and Event Types: `Idp Group Mapping – Create`, `Idp Group Mapping – Delete` and `Idp Group Mapping – Update`6. Verify that in the `Actions` section the Action Type contains: `Notifications` and that a valid `Topic` is referenced.**From CLI:** 1. Find the OCID of the specific Event Rule based on Display Name and Compartment OCID```oci events rule list --compartment-id <compartment-ocid> --query data [?\\display-name\\=='<displa-name>'].{id:id} --output table```2. List the details of a specific Event Rule based on the OCID of the rule.```oci events rule get --rule-id <rule-id>```3. In the JSON output locate the Conditions key value pair and verify that the following Conditions are present:```com.oraclecloud.identitycontrolplane.addidpgroupmappingcom.oraclecloud.identitycontrolplane.removeidpgroupmappingcom.oraclecloud.identitycontrolplane.updateidpgroupmapping```4. Verify the value of the `is-enabled` attribute is `true`5. In the JSON output verify that `actionType` is `ONS` and locate the `topic-id`6. Verify the correct topic is used by checking the topic name```oci ons topic get --topic-id <topic-id> --query data.{name:name} --output table```",
+          "RemediationProcedure": "**From Console:**1. Go to the `Events Service` page: [https://cloud.oracle.com/events/rules](https://cloud.oracle.com/events/rules)2. Select the `compartment` that should host the rule3. Click `Create Rule`4. Provide a `Display Name` and `Description`5. Create a Rule Condition by selecting `Identity` in the Service Name Drop-down and selecting `Idp Group Mapping – Add`, `Idp Group Mapping – Remove` and `Idp Group Mapping – Update`6. In the `Actions` section select `Notifications` as Action Type7. Select the `Compartment` that hosts the Topic to be used.8. Select the `Topic` to be used9. Optionally add Tags to the Rule10. Click `Create Rule`**From CLI:**1. Find the `topic-id` of the topic the Event Rule should use for sending notifications by using the topic `name` and `Compartment OCID````oci ons topic list --compartment-id <compartment-ocid> --all --query data [?name=='<topic-name>'].{name:name,topic_id:\\topic-id\\} --output table```2. Create a JSON file to be used when creating the Event Rule. Replace topic id, display name, description and compartment OCID.```{ actions: { actions: [ { actionType: ONS, isEnabled: true, topicId: <topic-id> }] }, condition:{\\eventType\\:[\\com.oraclecloud.identitycontrolplane.addidpgroupmapping\\,\\com.oraclecloud.identitycontrolplane.removeidpgroupmapping\\,\\com.oraclecloud.identitycontrolplane.updateidpgroupmapping\\],\\data\\:{}}, displayName: <display-name>, description: <description>, isEnabled: true, compartmentId: <compartment-ocid>}```3. Create the actual event rule```oci events rule create --from-json file://event_rule.json```4. Note in the JSON returned that it lists the parameters specified in the JSON file provided and that there is an OCID provided for the Event Rule",
+          "AuditProcedure": "**From Console:**1. Go to the Events Service page: [https://cloud.oracle.com/events/rules](https://cloud.oracle.com/events/rules)2. Select the `Compartment` that hosts the rules3. Find and click the `Rule` that handles `Idp Group Mapping` Changes (if any)4. Ensure the `Rule` is `ACTIVE`5. Click the `Edit Rule` button and verify that the `RuleConditions` section contains a condition for the Service `Identity` and Event Types: `Idp Group Mapping – Add`, `Idp Group Mapping – Remove` and `Idp Group Mapping – Update`6. Verify that in the `Actions` section the Action Type contains: `Notifications` and that a valid `Topic` is referenced.**From CLI:** 1. Find the OCID of the specific Event Rule based on Display Name and Compartment OCID```oci events rule list --compartment-id <compartment-ocid> --query data [?\\display-name\\=='<displa-name>'].{id:id} --output table```2. List the details of a specific Event Rule based on the OCID of the rule.```oci events rule get --rule-id <rule-id>```3. In the JSON output locate the Conditions key value pair and verify that the following Conditions are present:```com.oraclecloud.identitycontrolplane.addidpgroupmappingcom.oraclecloud.identitycontrolplane.removeidpgroupmappingcom.oraclecloud.identitycontrolplane.updateidpgroupmapping```4. Verify the value of the `is-enabled` attribute is `true`5. In the JSON output verify that `actionType` is `ONS` and locate the `topic-id`6. Verify the correct topic is used by checking the topic name```oci ons topic get --topic-id <topic-id> --query data.{name:name} --output table```",
           "AdditionalInformation": "'- The same Notification topic can be reused by many Event Rules.- The generated notification will include an eventID that can be used when querying the Audit Logs in case further investigation is required.",
           "References": ""
         }

prowler/config/config.py

@@ -38,7 +38,7 @@ class _MutableTimestamp:
 
 timestamp = _MutableTimestamp(datetime.today())
 timestamp_utc = _MutableTimestamp(datetime.now(timezone.utc))
-prowler_version = "5.22.0"
+prowler_version = "5.22.1"
 html_logo_url = "https://github.com/prowler-cloud/prowler/"
 square_logo_img = "https://raw.githubusercontent.com/prowler-cloud/prowler/dc7d2d5aeb92fdf12e8604f42ef6472cd3e8e889/docs/img/prowler-logo-black.png"
 aws_logo = "https://user-images.githubusercontent.com/38561120/235953920-3e3fba08-0795-41dc-b480-9bea57db9f2e.png"

prowler/providers/aws/aws_provider.py

@@ -942,20 +942,23 @@ class AwsProvider(Provider):
             )
             raise error
 
-    def get_default_region(self, service: str) -> str:
-        """get_default_region returns the default region based on the profile and audited service regions
+    def get_default_region(self, service: str, global_service: bool = False) -> str:
+        """get_default_region returns the default region based on the profile and audited service regions.
+
+        For global services (CloudFront, Route53, Shield, FMS) the partition's
+        global region is always returned, ignoring profile and audited regions.
 
         Args:
             - service: The AWS service name
+            - global_service: If True, return the partition's global region directly
 
         Returns:
             - str: The default region for the given service
-
-        Example:
-            service = "ec2"
-            default_region = get_default_region(service)
         """
         try:
+            if global_service:
+                return self.get_global_region()
+
             service_regions = AwsProvider.get_available_aws_service_regions(
                 service, self._identity.partition, self._identity.audited_regions
             )

prowler/providers/aws/lib/service/service.py

@@ -61,7 +61,9 @@ class AWSService:
         # Get a single region and client if the service needs it (e.g. AWS Global Service)
         # We cannot include this within an else because some services needs both the regional_clients
         # and a single client like S3
-        self.region = provider.get_default_region(self.service)
+        self.region = provider.get_default_region(
+            self.service, global_service=global_service
+        )
         self.client = self.session.client(self.service, self.region)
 
         # Thread pool for __threading_call__

prowler/providers/common/provider.py

@@ -321,7 +321,7 @@ class Provider(ABC):
                     provider_class(
                         oci_config_file=arguments.oci_config_file,
                         profile=arguments.profile,
-                        region=arguments.region,
+                        region=set(arguments.region) if arguments.region else None,
                         compartment_ids=arguments.compartment_id,
                         config_path=arguments.config_file,
                         mutelist_path=arguments.mutelist_file,

prowler/providers/oraclecloud/lib/arguments/arguments.py

@@ -44,8 +44,9 @@ def init_parser(self):
     oci_regions_subparser = oci_parser.add_argument_group("OCI Regions")
     oci_regions_subparser.add_argument(
         "--region",
-        "-r",
-        nargs="?",
+        "--filter-region",
+        "-f",
+        nargs="+",
         help="OCI region to run Prowler against. If not specified, all subscribed regions will be audited",
         choices=list(OCI_REGIONS.keys()),
     )

prowler/providers/oraclecloud/oraclecloud_provider.py

@@ -62,7 +62,7 @@ class OraclecloudProvider(Provider):
     _identity: OCIIdentityInfo
     _session: OCISession
     _audit_config: dict
-    _regions: list = []
+    _regions: set = set()
     _compartments: list = []
     _mutelist: OCIMutelist
     audit_metadata: Audit_Metadata
@@ -71,7 +71,7 @@ class OraclecloudProvider(Provider):
         self,
         oci_config_file: str = None,
         profile: str = None,
-        region: str = None,
+        region: set = set(),
         compartment_ids: list = None,
         config_path: str = None,
         config_content: dict = None,
@@ -92,7 +92,7 @@ class OraclecloudProvider(Provider):
         Args:
             - oci_config_file: The path to the OCI config file.
             - profile: The name of the OCI CLI profile to use.
-            - region: The OCI region to audit.
+            - region: The OCI region(s) to audit.
             - compartment_ids: A list of compartment OCIDs to audit.
             - config_path: The path to the Prowler configuration file.
             - config_content: The content of the configuration file.
@@ -127,6 +127,11 @@ class OraclecloudProvider(Provider):
 
         logger.info("Initializing OCI provider ...")
 
+        # Check if the configuration is scanning a single region
+        single_region = None
+        if region:
+            single_region = list(region)[0] if len(region) == 1 else None
+
         # Setup OCI Session
         logger.info("Setting up OCI session ...")
         self._session = self.setup_session(
@@ -138,7 +143,7 @@ class OraclecloudProvider(Provider):
             key_file=key_file,
             key_content=key_content,
             tenancy=tenancy,
-            region=region,
+            region=single_region,
             pass_phrase=pass_phrase,
         )
 
@@ -148,7 +153,7 @@ class OraclecloudProvider(Provider):
         logger.info("Validating OCI credentials ...")
         self._identity = self.set_identity(
             session=self._session,
-            region=region,
+            region=single_region,
             compartment_ids=compartment_ids,
         )
         logger.info("OCI credentials validated")
@@ -531,47 +536,42 @@ class OraclecloudProvider(Provider):
 
         return True
 
-    def get_regions_to_audit(self, region: str = None) -> list:
+    def get_regions_to_audit(self, region_set: set = None) -> list:
         """
         get_regions_to_audit returns the list of regions to audit.
 
         Args:
-            - region: The OCI region to audit.
+            - region: The OCI region(s) to audit.
 
         Returns:
             - list: The list of OCIRegion objects to audit.
         """
         regions = []
 
-        if region:
-            # Audit specific region
-            if region in OCI_REGIONS:
-                regions.append(
-                    OCIRegion(
-                        key=region,
-                        name=OCI_REGIONS[region],
-                        is_home_region=True,
-                    )
+        # Audit all subscribed regions
+        try:
+            # Create identity client with proper authentication handling
+            if self._session.signer:
+                identity_client = oci.identity.IdentityClient(
+                    config=self._session.config, signer=self._session.signer
                 )
             else:
-                logger.warning(f"Invalid region: {region}. Using default region.")
-        else:
-            # Audit all subscribed regions
-            try:
-                # Create identity client with proper authentication handling
-                if self._session.signer:
-                    identity_client = oci.identity.IdentityClient(
-                        config=self._session.config, signer=self._session.signer
-                    )
-                else:
-                    identity_client = oci.identity.IdentityClient(
-                        config=self._session.config
-                    )
-                region_subscriptions = identity_client.list_region_subscriptions(
-                    self._identity.tenancy_id
-                ).data
+                identity_client = oci.identity.IdentityClient(
+                    config=self._session.config
+                )
+            region_subscriptions = identity_client.list_region_subscriptions(
+                self._identity.tenancy_id
+            ).data
 
-                for region_sub in region_subscriptions:
+            # Check if auditing specific region or all
+            regions_check = (
+                region_set
+                if region_set
+                else [sub.region_name for sub in region_subscriptions]
+            )
+
+            for region_sub in region_subscriptions:
+                if region_sub.region_name in regions_check:
                     regions.append(
                         OCIRegion(
                             key=region_sub.region_name,
@@ -581,22 +581,20 @@ class OraclecloudProvider(Provider):
                             is_home_region=region_sub.is_home_region,
                         )
                     )
-
-                logger.info(f"Found {len(regions)} subscribed regions")
-
-            except Exception as error:
-                logger.warning(
-                    f"Could not retrieve region subscriptions: {error}. Using configured region."
-                )
-                # Fallback to configured region
-                config_region = self._session.config.get("region", "us-ashburn-1")
-                regions.append(
-                    OCIRegion(
-                        key=config_region,
-                        name=OCI_REGIONS.get(config_region, config_region),
-                        is_home_region=True,
-                    )
+            logger.info(f"Found {len(regions)} subscribed regions")
+        except Exception as error:
+            logger.warning(
+                f"Could not retrieve region subscriptions: {error}. Using configured region."
+            )
+            # Fallback to configured region
+            config_region = self._session.config.get("region", "us-ashburn-1")
+            regions.append(
+                OCIRegion(
+                    key=config_region,
+                    name=OCI_REGIONS.get(config_region, config_region),
+                    is_home_region=True,
                 )
+            )
 
         return regions
 

prowler/providers/oraclecloud/services/blockstorage/blockstorage_service.py

@@ -35,10 +35,9 @@ class BlockStorage(OCIService):
         Returns:
             Block Storage client instance
         """
-        client_region = self.regional_clients.get(region)
-        if client_region:
-            return self._create_oci_client(oci.core.BlockstorageClient)
-        return None
+        return self._create_oci_client(
+            oci.core.BlockstorageClient, config_overrides={"region": region}
+        )
 
     def __list_volumes__(self, regional_client):
         """
@@ -112,7 +111,8 @@ class BlockStorage(OCIService):
                 try:
                     # Get availability domains for this compartment
                     identity_client = self._create_oci_client(
-                        oci.identity.IdentityClient
+                        oci.identity.IdentityClient,
+                        config_overrides={"region": regional_client.region},
                     )
                     availability_domains = identity_client.list_availability_domains(
                         compartment_id=compartment.id

prowler/providers/oraclecloud/services/compute/compute_service.py

@@ -33,10 +33,9 @@ class Compute(OCIService):
         Returns:
             Compute client instance
         """
-        client_region = self.regional_clients.get(region)
-        if client_region:
-            return self._create_oci_client(oci.core.ComputeClient)
-        return None
+        return self._create_oci_client(
+            oci.core.ComputeClient, config_overrides={"region": region}
+        )
 
     def __list_instances__(self, regional_client):
         """

prowler/providers/oraclecloud/services/events/events_rule_idp_group_mapping_changes/events_rule_idp_group_mapping_changes.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "high",
   "ResourceType": "EventRule",
   "ResourceGroup": "messaging",
-  "Description": "**OCI Events rules** monitor **IdP group mapping changes** with **notification actions** for `com.oraclecloud.identitycontrolplane.createidpgroupmapping`, `com.oraclecloud.identitycontrolplane.deleteidpgroupmapping`, and `com.oraclecloud.identitycontrolplane.updateidpgroupmapping`.",
+  "Description": "**OCI Events rules** monitor **IdP group mapping changes** with **notification actions** for `com.oraclecloud.identitycontrolplane.addidpgroupmapping`, `com.oraclecloud.identitycontrolplane.removeidpgroupmapping`, and `com.oraclecloud.identitycontrolplane.updateidpgroupmapping`.",
   "Risk": "Without **alerts** on IdP group mapping changes, federated users can gain unauthorized group memberships unnoticed, enabling **privilege escalation** and broader access to OCI resources. This undermines **confidentiality** and **integrity**, and may affect **availability** through misuse of elevated permissions.",
   "RelatedUrl": "",
   "AdditionalURLs": [
@@ -18,13 +18,13 @@
   ],
   "Remediation": {
     "Code": {
-      "CLI": "oci events rule create --compartment-id <COMPARTMENT_OCID> --display-name <example_resource_name> --is-enabled true --condition '{\"eventType\":[\"com.oraclecloud.identitycontrolplane.createidpgroupmapping\",\"com.oraclecloud.identitycontrolplane.deleteidpgroupmapping\",\"com.oraclecloud.identitycontrolplane.updateidpgroupmapping\"]}' --actions '{\"actions\":[{\"actionType\":\"ONS\",\"isEnabled\":true,\"topicId\":\"<TOPIC_OCID>\"}]}'",
+      "CLI": "oci events rule create --compartment-id <COMPARTMENT_OCID> --display-name <example_resource_name> --is-enabled true --condition '{\"eventType\":[\"com.oraclecloud.identitycontrolplane.addidpgroupmapping\",\"com.oraclecloud.identitycontrolplane.removeidpgroupmapping\",\"com.oraclecloud.identitycontrolplane.updateidpgroupmapping\"]}' --actions '{\"actions\":[{\"actionType\":\"ONS\",\"isEnabled\":true,\"topicId\":\"<TOPIC_OCID>\"}]}'",
       "NativeIaC": "",
-      "Other": "1. In OCI Console, go to Observability & Management > Events Service > Rules\n2. Click Create rule\n3. Condition: add Event types:\n   - com.oraclecloud.identitycontrolplane.createidpgroupmapping\n   - com.oraclecloud.identitycontrolplane.deleteidpgroupmapping\n   - com.oraclecloud.identitycontrolplane.updateidpgroupmapping\n4. Actions: Add action > Notifications (ONS) and select the target Topic\n5. Ensure Rule is Enabled and click Create",
-      "Terraform": "```hcl\nresource \"oci_events_rule\" \"<example_resource_name>\" {\n  compartment_id = \"<example_resource_id>\"\n  display_name   = \"<example_resource_name>\"\n  is_enabled     = true\n\n  condition = jsonencode({\n    eventType = [\n      \"com.oraclecloud.identitycontrolplane.createidpgroupmapping\", # critical: monitor IdP group mapping create\n      \"com.oraclecloud.identitycontrolplane.deleteidpgroupmapping\", # critical: monitor IdP group mapping delete\n      \"com.oraclecloud.identitycontrolplane.updateidpgroupmapping\"  # critical: monitor IdP group mapping update\n    ]\n  })\n\n  actions {\n    actions {\n      action_type = \"ONS\"                       # critical: adds notification action\n      topic_id    = \"<example_resource_id>\"     # critical: ONS topic to notify\n      is_enabled  = true\n    }\n  }\n}\n```"
+      "Other": "1. In OCI Console, go to Observability & Management > Events Service > Rules\n2. Click Create rule\n3. Condition: add Event types:\n   - com.oraclecloud.identitycontrolplane.addidpgroupmapping\n   - com.oraclecloud.identitycontrolplane.removeidpgroupmapping\n   - com.oraclecloud.identitycontrolplane.updateidpgroupmapping\n4. Actions: Add action > Notifications (ONS) and select the target Topic\n5. Ensure Rule is Enabled and click Create",
+      "Terraform": "```hcl\nresource \"oci_events_rule\" \"<example_resource_name>\" {\n  compartment_id = \"<example_resource_id>\"\n  display_name   = \"<example_resource_name>\"\n  is_enabled     = true\n\n  condition = jsonencode({\n    eventType = [\n      \"com.oraclecloud.identitycontrolplane.addidpgroupmapping\", # critical: monitor IdP group mapping add\n      \"com.oraclecloud.identitycontrolplane.removeidpgroupmapping\", # critical: monitor IdP group mapping remove\n      \"com.oraclecloud.identitycontrolplane.updateidpgroupmapping\"  # critical: monitor IdP group mapping update\n    ]\n  })\n\n  actions {\n    actions {\n      action_type = \"ONS\"                       # critical: adds notification action\n      topic_id    = \"<example_resource_id>\"     # critical: ONS topic to notify\n      is_enabled  = true\n    }\n  }\n}\n```"
     },
     "Recommendation": {
-      "Text": "Define **Events rules** for IdP group mapping changes (`com.oraclecloud.identitycontrolplane.createidpgroupmapping`, `...deleteidpgroupmapping`, `...updateidpgroupmapping`) and route notifications to monitored channels via **OCI Notifications**. Apply **least privilege** and **separation of duties**, and integrate alerts with a SIEM for **defense in depth**.",
+      "Text": "Define **Events rules** for IdP group mapping changes (`com.oraclecloud.identitycontrolplane.addidpgroupmapping`, `...removeidpgroupmapping`, `...updateidpgroupmapping`) and route notifications to monitored channels via **OCI Notifications**. Apply **least privilege** and **separation of duties**, and integrate alerts with a SIEM for **defense in depth**.",
       "Url": "https://hub.prowler.com/check/events_rule_idp_group_mapping_changes"
     }
   },

prowler/providers/oraclecloud/services/events/events_rule_idp_group_mapping_changes/events_rule_idp_group_mapping_changes.py

@@ -15,17 +15,33 @@ class events_rule_idp_group_mapping_changes(Check):
         """Execute the events_rule_idp_group_mapping_changes check."""
         findings = []
 
-        # Required event types for IdP group mapping changes
-        required_event_types = [
+        # OCI CIS 3.1 renamed create/delete to add/remove. Accept both to keep
+        # compatibility with rules that still use the legacy event names.
+        current_required_event_types = [
+            "com.oraclecloud.identitycontrolplane.addidpgroupmapping",
+            "com.oraclecloud.identitycontrolplane.removeidpgroupmapping",
+            "com.oraclecloud.identitycontrolplane.updateidpgroupmapping",
+        ]
+        legacy_required_event_types = [
             "com.oraclecloud.identitycontrolplane.createidpgroupmapping",
             "com.oraclecloud.identitycontrolplane.deleteidpgroupmapping",
             "com.oraclecloud.identitycontrolplane.updateidpgroupmapping",
         ]
 
         # Filter rules that monitor IdP group mapping changes
-        matching_rules = filter_rules_by_event_types(
-            events_client.rules, required_event_types
-        )
+        matching_rules = []
+        seen_rule_ids = set()
+        for required_event_types in (
+            current_required_event_types,
+            legacy_required_event_types,
+        ):
+            for rule, condition in filter_rules_by_event_types(
+                events_client.rules, required_event_types
+            ):
+                rule_id = getattr(rule, "id", None) or id(rule)
+                if rule_id not in seen_rule_ids:
+                    matching_rules.append((rule, condition))
+                    seen_rule_ids.add(rule_id)
 
         # Create findings for each matching rule
         for rule, _ in matching_rules:

prowler/providers/oraclecloud/services/filestorage/filestorage_service.py

@@ -20,10 +20,9 @@ class Filestorage(OCIService):
 
     def __get_client__(self, region):
         """Get the Filestorage client for a region."""
-        client_region = self.regional_clients.get(region)
-        if client_region:
-            return self._create_oci_client(oci.file_storage.FileStorageClient)
-        return None
+        return self._create_oci_client(
+            oci.file_storage.FileStorageClient, config_overrides={"region": region}
+        )
 
     def __list_file_systems__(self, regional_client):
         """List all file_systems."""
@@ -40,7 +39,8 @@ class Filestorage(OCIService):
                 try:
                     # Get availability domains for this compartment
                     identity_client = self._create_oci_client(
-                        oci.identity.IdentityClient
+                        oci.identity.IdentityClient,
+                        config_overrides={"region": regional_client.region},
                     )
                     availability_domains = identity_client.list_availability_domains(
                         compartment_id=compartment.id

prowler/providers/oraclecloud/services/identity/identity_password_policy_expires_within_365_days/identity_password_policy_expires_within_365_days.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "Policy",
   "ResourceGroup": "IAM",
-  "Description": "**OCI Identity Domain password policies** are evaluated to confirm **password expiration** is configured and set to `<= 365` days (`password_expires_after`).\n\n*Legacy IAM lacks password expiration; tenancies without Identity Domains require manual assessment.*",
+  "Description": "**OCI Identity Domain password policies** are evaluated to confirm **password expiration** is configured and set to `<= 365` days (`password_expires_after`). System-managed immutable policies (`SimplePasswordPolicy`, `StandardPasswordPolicy`) are excluded.\n\n*Legacy IAM lacks password expiration; tenancies without Identity Domains require manual assessment.*",
   "Risk": "Missing or >`365`-day **password expiration** extends the window for **credential stuffing**, **brute force**, and use of leaked passwords. This enables unauthorized access, data exposure, and configuration changes, harming **confidentiality** and **integrity**, and allowing attacker persistence after staff turnover.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/oraclecloud/services/identity/identity_password_policy_minimum_length_14/identity_password_policy_minimum_length_14.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "high",
   "ResourceType": "Policy",
   "ResourceGroup": "IAM",
-  "Description": "**OCI IAM password policies** are evaluated to confirm a **minimum password length** of `>= 14` characters is enforced. The assessment considers policies defined in **Identity Domains** and the legacy tenancy policy, and also detects when no password policy exists.",
+  "Description": "**OCI IAM password policies** are evaluated to confirm a **minimum password length** of `>= 14` characters is enforced. The assessment considers policies defined in **Identity Domains** and the legacy tenancy policy, and also detects when no password policy exists. System-managed immutable policies (`SimplePasswordPolicy`, `StandardPasswordPolicy`) are excluded.",
   "Risk": "Short or missing password requirements weaken authentication, enabling **brute-force**, **password spraying**, and faster **offline cracking**. Compromised accounts can enable unauthorized console/API use, leading to **data exfiltration** (C), **unauthorized changes** (I), and service disruption via destructive actions (A).",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/oraclecloud/services/identity/identity_password_policy_prevents_reuse/identity_password_policy_prevents_reuse.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "User",
   "ResourceGroup": "IAM",
-  "Description": "**OCI Identity Domains** password policies are evaluated for **password reuse prevention** via the **password history** setting. The finding expects a configured policy with `num_passwords_in_history` set to at least `24`. *Legacy IAM password policies lack password history; only Identity Domains support this setting.*",
+  "Description": "**OCI Identity Domains** password policies are evaluated for **password reuse prevention** via **password history** (`num_passwords_in_history >= 24`). System-managed policies (`SimplePasswordPolicy`, `StandardPasswordPolicy`) are excluded. *Legacy IAM lacks password history.*",
   "Risk": "Without **password history**, users can reuse old passwords. Compromised credentials remain valid after resets, enabling account takeover, unauthorized changes, and **lateral movement**, degrading **confidentiality** and **integrity** and weakening recovery from password-related incidents.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/oraclecloud/services/identity/identity_service.py

@@ -35,7 +35,7 @@ class Identity(OCIService):
         self.__threading_call__(self.__list_dynamic_groups__)
         self.__threading_call__(self.__list_domains__)
         self.__threading_call__(self.__list_domain_password_policies__)
-        self.__get_password_policy__()
+        self.__threading_call__(self.__get_password_policy__)
         self.__threading_call__(self.__search_root_compartment_resources__)
         self.__threading_call__(self.__search_active_non_root_compartments__)
 
@@ -49,10 +49,9 @@ class Identity(OCIService):
         Returns:
             Identity client instance
         """
-        client_region = self.regional_clients.get(region)
-        if client_region:
-            return self._create_oci_client(oci.identity.IdentityClient)
-        return None
+        return self._create_oci_client(
+            oci.identity.IdentityClient, config_overrides={"region": region}
+        )
 
     def __list_users__(self, regional_client):
         """
@@ -66,7 +65,7 @@ class Identity(OCIService):
             if regional_client.region not in self.provider.identity.region:
                 return
 
-            identity_client = self._create_oci_client(oci.identity.IdentityClient)
+            identity_client = self.__get_client__(regional_client.region)
 
             logger.info("Identity - Listing Users...")
 
@@ -316,7 +315,7 @@ class Identity(OCIService):
             if regional_client.region not in self.provider.identity.region:
                 return
 
-            identity_client = self._create_oci_client(oci.identity.IdentityClient)
+            identity_client = self.__get_client__(regional_client.region)
 
             logger.info("Identity - Listing Groups...")
 
@@ -359,7 +358,7 @@ class Identity(OCIService):
             if regional_client.region not in self.provider.identity.region:
                 return
 
-            identity_client = self._create_oci_client(oci.identity.IdentityClient)
+            identity_client = self.__get_client__(regional_client.region)
 
             logger.info("Identity - Listing Policies...")
 
@@ -404,7 +403,7 @@ class Identity(OCIService):
             if regional_client.region not in self.provider.identity.region:
                 return
 
-            identity_client = self._create_oci_client(oci.identity.IdentityClient)
+            identity_client = self.__get_client__(regional_client.region)
 
             logger.info("Identity - Listing Dynamic Groups...")
 
@@ -452,7 +451,7 @@ class Identity(OCIService):
             if regional_client.region not in self.provider.identity.region:
                 return
 
-            identity_client = self._create_oci_client(oci.identity.IdentityClient)
+            identity_client = self.__get_client__(regional_client.region)
 
             logger.info("Identity - Listing Identity Domains...")
 
@@ -518,6 +517,14 @@ class Identity(OCIService):
                     policies_response = domain_client.list_password_policies()
 
                     for policy in policies_response.data.resources:
+                        # Skip system-managed immutable policies that are
+                        # hidden in the OCI Console and not user-configurable
+                        if policy.id in (
+                            "SimplePasswordPolicy",
+                            "StandardPasswordPolicy",
+                        ):
+                            continue
+
                         domain.password_policies.append(
                             DomainPasswordPolicy(
                                 id=policy.id,
@@ -541,10 +548,13 @@ class Identity(OCIService):
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
-    def __get_password_policy__(self):
+    def __get_password_policy__(self, regional_client):
         """Get the password policy for the tenancy."""
         try:
-            identity_client = self._create_oci_client(oci.identity.IdentityClient)
+            if regional_client.region not in self.provider.identity.region:
+                return
+
+            identity_client = self.__get_client__(regional_client.region)
 
             logger.info("Identity - Getting Password Policy...")
 
@@ -576,7 +586,8 @@ class Identity(OCIService):
 
             # Create search client using the helper method for proper authentication
             search_client = self._create_oci_client(
-                oci.resource_search.ResourceSearchClient
+                oci.resource_search.ResourceSearchClient,
+                config_overrides={"region": regional_client.region},
             )
 
             # Query to search for resources in root compartment
@@ -623,7 +634,8 @@ class Identity(OCIService):
 
             # Create search client using the helper method for proper authentication
             search_client = self._create_oci_client(
-                oci.resource_search.ResourceSearchClient
+                oci.resource_search.ResourceSearchClient,
+                config_overrides={"region": regional_client.region},
             )
 
             # Query to search for active compartments in the tenancy (excluding root)

prowler/providers/oraclecloud/services/kms/kms_key_rotation_enabled/kms_key_rotation_enabled.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "high",
   "ResourceType": "Key",
   "ResourceGroup": "security",
-  "Description": "**OCI KMS customer-managed keys** configured for **automatic rotation** or with a rotation interval set to `<= 365` days.",
+  "Description": "**OCI KMS customer-managed keys** configured for **automatic rotation**, with a rotation interval set to `<= 365` days, or **manually rotated** within the last 365 days. Some vault types do not support auto-rotation, so manual rotation is accepted as an alternative.",
   "Risk": "Without regular rotation, a compromised key can be used longer to decrypt data at rest and backups or to forge signatures. This erodes **confidentiality** and **integrity**, increases the blast radius, and complicates incident response due to broad reuse of the same key version.",
   "RelatedUrl": "",
   "AdditionalURLs": [
@@ -20,11 +20,11 @@
     "Code": {
       "CLI": "oci kms management key update --key-id <example_resource_id> --endpoint <example_management_endpoint> --is-auto-rotation-enabled true --auto-key-rotation-details '{\"rotationIntervalInDays\": 365}'",
       "NativeIaC": "",
-      "Other": "1. In OCI Console, go to Identity & Security > Vault\n2. Open the vault, then under Resources select Master Encryption Keys\n3. Click the target key name\n4. Click Edit auto-rotation settings\n5. Enable Auto rotation and set Rotation interval to 365 days (or less)\n6. Click Update",
+      "Other": "1. In OCI Console, go to Identity & Security > Vault\n2. Open the vault, then under Resources select Master Encryption Keys\n3. Click the target key name\n4. For vaults that support auto-rotation: Click Edit auto-rotation settings, enable Auto rotation, set Rotation interval to 365 days (or less), and click Update\n5. For vaults that do not support auto-rotation (e.g., External or Virtual Private vaults): Click 'Rotate Key' to manually rotate the key version at least once every 365 days",
       "Terraform": "```hcl\nresource \"oci_kms_key\" \"<example_resource_name>\" {\n  compartment_id      = \"<example_resource_id>\"\n  display_name        = \"<example_resource_name>\"\n  management_endpoint = \"<example_management_endpoint>\"\n\n  key_shape {\n    algorithm = \"AES\"\n    length    = 16\n  }\n\n  is_auto_rotation_enabled = true  # Critical: enables auto rotation\n  auto_key_rotation_details {\n    rotation_interval_in_days = 365  # Critical: interval <= 365 days\n  }\n}\n```"
     },
     "Recommendation": {
-      "Text": "Enable **automatic key rotation** and set an interval `<= 365` days (*shorter for sensitive data*). Apply **least privilege** and **separation of duties** for key administration. Monitor rotation status, retire old key versions, and ensure applications handle key versioning to prevent outages.",
+      "Text": "Enable **automatic key rotation** and set an interval `<= 365` days (*shorter for sensitive data*). For vault types that do not support auto-rotation (e.g., External or Virtual Private vaults), **manually rotate** the key at least once every 365 days. Apply **least privilege** and **separation of duties** for key administration. Monitor rotation status, retire old key versions, and ensure applications handle key versioning to prevent outages.",
       "Url": "https://hub.prowler.com/check/kms_key_rotation_enabled"
     }
   },

prowler/providers/oraclecloud/services/kms/kms_key_rotation_enabled/kms_key_rotation_enabled.py

@@ -1,5 +1,7 @@
 """Check Ensure customer created Customer Managed Key (CMK) is rotated at least annually."""
 
+from datetime import datetime, timedelta, timezone
+
 from prowler.lib.check.models import Check, Check_Report_OCI
 from prowler.providers.oraclecloud.services.kms.kms_client import kms_client
 
@@ -21,16 +23,32 @@ class kms_key_rotation_enabled(Check):
                 compartment_id=key.compartment_id,
             )
 
-            # Check if auto-rotation is enabled OR if rotation interval is set and <= 365 days
-            if key.is_auto_rotation_enabled or (
-                key.rotation_interval_in_days is not None
-                and key.rotation_interval_in_days <= 365
+            now = datetime.now(timezone.utc)
+            max_age = timedelta(days=365)
+
+            manually_rotated_recently = (
+                key.current_key_version_time_created is not None
+                and (now - key.current_key_version_time_created) <= max_age
+            )
+
+            if (
+                key.is_auto_rotation_enabled
+                or (
+                    key.rotation_interval_in_days is not None
+                    and key.rotation_interval_in_days <= 365
+                )
+                or manually_rotated_recently
             ):
                 report.status = "PASS"
-                report.status_extended = f"KMS key '{key.name}' has rotation enabled (auto-rotation: {key.is_auto_rotation_enabled}, interval: {key.rotation_interval_in_days} days)."
+                if key.is_auto_rotation_enabled:
+                    report.status_extended = f"KMS key {key.name} has auto-rotation enabled with interval of {key.rotation_interval_in_days} days."
+                elif manually_rotated_recently:
+                    report.status_extended = f"KMS key {key.name} was manually rotated within the last 365 days."
+                else:
+                    report.status_extended = f"KMS key {key.name} has rotation interval set to {key.rotation_interval_in_days} days."
             else:
                 report.status = "FAIL"
-                report.status_extended = f"KMS key '{key.name}' does not have rotation enabled or rotation interval exceeds 365 days."
+                report.status_extended = f"KMS key {key.name} has not been rotated within the last 365 days and does not have auto-rotation enabled."
 
             findings.append(report)
 

prowler/providers/oraclecloud/services/kms/kms_service.py

@@ -1,5 +1,6 @@
 """OCI Kms Service Module."""
 
+from datetime import datetime
 from typing import Optional
 
 import oci
@@ -20,10 +21,9 @@ class Kms(OCIService):
 
     def __get_client__(self, region):
         """Get the Kms client for a region."""
-        client_region = self.regional_clients.get(region)
-        if client_region:
-            return self._create_oci_client(oci.key_management.KmsVaultClient)
-        return None
+        return self._create_oci_client(
+            oci.key_management.KmsVaultClient, config_overrides={"region": region}
+        )
 
     def __list_keys__(self, regional_client):
         """List all keys."""
@@ -78,6 +78,25 @@ class Kms(OCIService):
                                         key_id=key_summary.id
                                     ).data
 
+                                    # Fetch current key version to get its creation time
+                                    current_key_version_time_created = None
+                                    if (
+                                        hasattr(key_details, "current_key_version")
+                                        and key_details.current_key_version
+                                    ):
+                                        try:
+                                            key_version = kms_management_client.get_key_version(
+                                                key_id=key_details.id,
+                                                key_version_id=key_details.current_key_version,
+                                            ).data
+                                            current_key_version_time_created = (
+                                                key_version.time_created
+                                            )
+                                        except Exception as version_error:
+                                            logger.warning(
+                                                f"Could not fetch key version for {key_details.id}: {version_error}"
+                                            )
+
                                     self.keys.append(
                                         Key(
                                             id=key_details.id,
@@ -110,6 +129,7 @@ class Kms(OCIService):
                                                 )
                                                 else None
                                             ),
+                                            current_key_version_time_created=current_key_version_time_created,
                                         )
                                     )
                 except Exception as error:
@@ -134,3 +154,4 @@ class Key(BaseModel):
     lifecycle_state: str
     is_auto_rotation_enabled: bool = False
     rotation_interval_in_days: Optional[int] = None
+    current_key_version_time_created: Optional[datetime] = None

pyproject.toml

@@ -94,7 +94,7 @@ maintainers = [{name = "Prowler Engineering", email = "engineering@prowler.com"}
 name = "prowler"
 readme = "README.md"
 requires-python = ">3.9.1,<3.13"
-version = "5.22.0"
+version = "5.22.1"
 
 [project.scripts]
 prowler = "prowler.__main__:prowler"

tests/providers/aws/aws_provider_test.py

@@ -943,6 +943,68 @@ aws:
         aws_provider._identity.profile_region = "non-existent-region"
         assert aws_provider.get_default_region("ec2") == AWS_REGION_EU_WEST_1
 
+    @mock_aws
+    def test_get_default_region_global_service_ignores_profile_region(self):
+        region = [AWS_REGION_EU_WEST_1]
+        aws_provider = AwsProvider(
+            regions=region,
+        )
+        aws_provider._identity.profile_region = AWS_REGION_EU_WEST_1
+
+        assert (
+            aws_provider.get_default_region("cloudfront", global_service=True)
+            == AWS_REGION_US_EAST_1
+        )
+
+    @mock_aws
+    def test_get_default_region_global_service_ignores_audited_regions(self):
+        region = [AWS_REGION_EU_WEST_1]
+        aws_provider = AwsProvider(
+            regions=region,
+        )
+        aws_provider._identity.profile_region = None
+
+        assert (
+            aws_provider.get_default_region("route53", global_service=True)
+            == AWS_REGION_US_EAST_1
+        )
+
+    @mock_aws
+    def test_get_default_region_global_service_china_partition(self):
+        aws_provider = AwsProvider()
+        aws_provider._identity.partition = AWS_CHINA_PARTITION
+        aws_provider._identity.profile_region = AWS_REGION_CN_NORTHWEST_1
+
+        assert (
+            aws_provider.get_default_region("cloudfront", global_service=True)
+            == AWS_REGION_CN_NORTH_1
+        )
+
+    @mock_aws
+    def test_get_default_region_global_service_gov_cloud_partition(self):
+        aws_provider = AwsProvider()
+        aws_provider._identity.partition = AWS_GOV_CLOUD_PARTITION
+        aws_provider._identity.profile_region = "us-gov-west-1"
+
+        assert (
+            aws_provider.get_default_region("shield", global_service=True)
+            == AWS_REGION_GOV_CLOUD_US_EAST_1
+        )
+
+    @mock_aws
+    def test_get_default_region_non_global_service_unaffected(self):
+        """Ensure global_service=False (default) still follows profile region logic."""
+        region = [AWS_REGION_EU_WEST_1]
+        aws_provider = AwsProvider(
+            regions=region,
+        )
+        aws_provider._identity.profile_region = AWS_REGION_EU_WEST_1
+
+        assert (
+            aws_provider.get_default_region("ec2", global_service=False)
+            == AWS_REGION_EU_WEST_1
+        )
+
     @mock_aws
     def test_aws_gov_get_global_region(self):
         aws_provider = AwsProvider()

tests/providers/aws/lib/service/service_test.py

@@ -4,7 +4,13 @@ from prowler.providers.aws.lib.service.service import AWSService
 from tests.providers.aws.utils import (
     AWS_ACCOUNT_ARN,
     AWS_ACCOUNT_NUMBER,
+    AWS_CHINA_PARTITION,
     AWS_COMMERCIAL_PARTITION,
+    AWS_GOV_CLOUD_PARTITION,
+    AWS_REGION_CN_NORTH_1,
+    AWS_REGION_CN_NORTHWEST_1,
+    AWS_REGION_EU_WEST_1,
+    AWS_REGION_GOV_CLOUD_US_EAST_1,
     AWS_REGION_US_EAST_1,
     set_mocked_aws_provider,
 )
@@ -61,6 +67,44 @@ class TestAWSService:
         assert service.region == AWS_REGION_US_EAST_1
         assert service.client.__class__.__name__ == "CloudFront"
 
+    def test_AWSService_global_service_uses_global_region_with_profile_region(self):
+        """Global services must use the partition's global region, not the profile region."""
+        service_name = "cloudfront"
+        provider = set_mocked_aws_provider(profile_region=AWS_REGION_EU_WEST_1)
+        service = AWSService(service_name, provider, global_service=True)
+
+        assert service.region == AWS_REGION_US_EAST_1
+
+    def test_AWSService_non_global_service_uses_profile_region(self):
+        """Non-global services should use the profile region when available."""
+        service_name = "s3"
+        provider = set_mocked_aws_provider(profile_region=AWS_REGION_EU_WEST_1)
+        service = AWSService(service_name, provider)
+
+        assert service.region == AWS_REGION_EU_WEST_1
+
+    def test_AWSService_global_service_china_partition(self):
+        """Global services in aws-cn partition should use cn-north-1."""
+        service_name = "cloudfront"
+        provider = set_mocked_aws_provider(
+            audited_partition=AWS_CHINA_PARTITION,
+            profile_region=AWS_REGION_CN_NORTHWEST_1,
+        )
+        service = AWSService(service_name, provider, global_service=True)
+
+        assert service.region == AWS_REGION_CN_NORTH_1
+
+    def test_AWSService_global_service_gov_cloud_partition(self):
+        """Global services in aws-us-gov partition should use us-gov-east-1."""
+        service_name = "cloudfront"
+        provider = set_mocked_aws_provider(
+            audited_partition=AWS_GOV_CLOUD_PARTITION,
+            profile_region="us-gov-west-1",
+        )
+        service = AWSService(service_name, provider, global_service=True)
+
+        assert service.region == AWS_REGION_GOV_CLOUD_US_EAST_1
+
     def test_AWSService_set_failed_check(self):
 
         AWSService.failed_checks.clear()

tests/providers/oraclecloud/services/events/events_rule_idp_group_mapping_changes/events_rule_idp_group_mapping_changes_test.py

@@ -9,6 +9,102 @@ from tests.providers.oraclecloud.oci_fixtures import (
 
 
 class Test_events_rule_idp_group_mapping_changes:
+    def test_current_cis_3_1_event_types_pass(self):
+        """events_rule_idp_group_mapping_changes: current CIS 3.1 event types should pass."""
+        events_client = mock.MagicMock()
+        events_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
+        events_client.audited_tenancy = OCI_TENANCY_ID
+
+        rule = mock.MagicMock()
+        rule.id = "ocid1.eventrule.oc1.iad.aaaaaaaexample"
+        rule.name = "idp-group-mapping-events"
+        rule.region = OCI_REGION
+        rule.compartment_id = OCI_COMPARTMENT_ID
+        rule.lifecycle_state = "ACTIVE"
+        rule.is_enabled = True
+        rule.condition = """
+        {
+          "eventType": [
+            "com.oraclecloud.identitycontrolplane.addidpgroupmapping",
+            "com.oraclecloud.identitycontrolplane.removeidpgroupmapping",
+            "com.oraclecloud.identitycontrolplane.updateidpgroupmapping"
+          ]
+        }
+        """
+        rule.actions = [{"action_type": "ONS", "is_enabled": True}]
+
+        events_client.rules = [rule]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.events.events_rule_idp_group_mapping_changes.events_rule_idp_group_mapping_changes.events_client",
+                new=events_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.events.events_rule_idp_group_mapping_changes.events_rule_idp_group_mapping_changes import (
+                events_rule_idp_group_mapping_changes,
+            )
+
+            check = events_rule_idp_group_mapping_changes()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == rule.id
+            assert result[0].resource_name == rule.name
+
+    def test_legacy_event_types_still_pass(self):
+        """events_rule_idp_group_mapping_changes: legacy event types remain supported."""
+        events_client = mock.MagicMock()
+        events_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
+        events_client.audited_tenancy = OCI_TENANCY_ID
+
+        rule = mock.MagicMock()
+        rule.id = "ocid1.eventrule.oc1.iad.bbbbbbbexample"
+        rule.name = "legacy-idp-group-mapping-events"
+        rule.region = OCI_REGION
+        rule.compartment_id = OCI_COMPARTMENT_ID
+        rule.lifecycle_state = "ACTIVE"
+        rule.is_enabled = True
+        rule.condition = """
+        {
+          "eventType": [
+            "com.oraclecloud.identitycontrolplane.createidpgroupmapping",
+            "com.oraclecloud.identitycontrolplane.deleteidpgroupmapping",
+            "com.oraclecloud.identitycontrolplane.updateidpgroupmapping"
+          ]
+        }
+        """
+        rule.actions = [{"action_type": "ONS", "is_enabled": True}]
+
+        events_client.rules = [rule]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.events.events_rule_idp_group_mapping_changes.events_rule_idp_group_mapping_changes.events_client",
+                new=events_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.events.events_rule_idp_group_mapping_changes.events_rule_idp_group_mapping_changes import (
+                events_rule_idp_group_mapping_changes,
+            )
+
+            check = events_rule_idp_group_mapping_changes()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == rule.id
+            assert result[0].resource_name == rule.name
+
     def test_no_resources(self):
         """events_rule_idp_group_mapping_changes: No resources to check"""
         events_client = mock.MagicMock()

tests/providers/oraclecloud/services/identity/identity_password_policy_expires_within_365_days/identity_password_policy_expires_within_365_days_test.py

@@ -1,5 +1,10 @@
+from datetime import datetime, timezone
 from unittest import mock
 
+from prowler.providers.oraclecloud.services.identity.identity_service import (
+    DomainPasswordPolicy,
+    IdentityDomain,
+)
 from tests.providers.oraclecloud.oci_fixtures import (
     OCI_COMPARTMENT_ID,
     OCI_REGION,
@@ -7,36 +12,34 @@ from tests.providers.oraclecloud.oci_fixtures import (
     set_mocked_oraclecloud_provider,
 )
 
+DOMAIN_ID = "ocid1.domain.oc1..aaaaaaaexample"
+DOMAIN_NAME = "Default"
+DOMAIN_URL = "https://idcs-example.identity.oraclecloud.com"
+POLICY_ID = "ocid1.passwordpolicy.oc1..aaaaaaaexample"
+POLICY_NAME = "CustomPasswordPolicy"
+
+
+def _make_domain(password_policies=None):
+    return IdentityDomain(
+        id=DOMAIN_ID,
+        display_name=DOMAIN_NAME,
+        description="Default identity domain",
+        url=DOMAIN_URL,
+        home_region=OCI_REGION,
+        compartment_id=OCI_COMPARTMENT_ID,
+        lifecycle_state="ACTIVE",
+        time_created=datetime.now(timezone.utc),
+        region=OCI_REGION,
+        password_policies=password_policies or [],
+    )
+
 
 class Test_identity_password_policy_expires_within_365_days:
-    def test_no_resources(self):
-        """identity_password_policy_expires_within_365_days: No resources to check"""
+    def test_no_domains(self):
+        """No Identity Domains → MANUAL finding."""
         identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock empty collections
-        identity_client.rules = []
-        identity_client.topics = []
-        identity_client.subscriptions = []
-        identity_client.users = []
-        identity_client.groups = []
-        identity_client.policies = []
-        identity_client.compartments = []
-        identity_client.instances = []
-        identity_client.volumes = []
-        identity_client.boot_volumes = []
-        identity_client.buckets = []
-        identity_client.keys = []
-        identity_client.file_systems = []
-        identity_client.databases = []
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.subnets = []
-        identity_client.vcns = []
-        identity_client.configuration = None
-        identity_client.active_non_root_compartments = []
-        identity_client.password_policy = None
+        identity_client.domains = []
 
         with (
             mock.patch(
@@ -55,134 +58,29 @@ class Test_identity_password_policy_expires_within_365_days:
             check = identity_password_policy_expires_within_365_days()
             result = check.execute()
 
-            # Verify result is a list (empty or with findings)
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "MANUAL"
 
-    def test_resource_compliant(self):
-        """identity_password_policy_expires_within_365_days: Resource passes the check (PASS)"""
+    def test_policy_expires_within_365_days(self):
+        """Password expires within 365 days → PASS."""
         identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock a compliant resource
-        resource = mock.MagicMock()
-        resource.id = "ocid1.resource.oc1.iad.aaaaaaaexample"
-        resource.name = "compliant-resource"
-        resource.region = OCI_REGION
-        resource.compartment_id = OCI_COMPARTMENT_ID
-        resource.lifecycle_state = "ACTIVE"
-        resource.tags = {"Environment": "Production"}
-
-        # Set attributes that make the resource compliant
-        resource.versioning = "Enabled"
-        resource.is_auto_rotation_enabled = True
-        resource.rotation_interval_in_days = 90
-        resource.public_access_type = "NoPublicAccess"
-        resource.logging_enabled = True
-        resource.kms_key_id = "ocid1.key.oc1.iad.aaaaaaaexample"
-        resource.in_transit_encryption = "ENABLED"
-        resource.is_secure_boot_enabled = True
-        resource.legacy_endpoint_disabled = True
-        resource.is_legacy_imds_endpoint_disabled = True
-
-        # Mock client with compliant resource
-        identity_client.buckets = [resource]
-        identity_client.keys = [resource]
-        identity_client.volumes = [resource]
-        identity_client.boot_volumes = [resource]
-        identity_client.instances = [resource]
-        identity_client.file_systems = [resource]
-        identity_client.databases = [resource]
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.rules = []
-        identity_client.configuration = resource
-        identity_client.users = []
-
-        with (
-            mock.patch(
-                "prowler.providers.common.provider.Provider.get_global_provider",
-                return_value=set_mocked_oraclecloud_provider(),
-            ),
-            mock.patch(
-                "prowler.providers.oraclecloud.services.identity.identity_password_policy_expires_within_365_days.identity_password_policy_expires_within_365_days.identity_client",
-                new=identity_client,
-            ),
-        ):
-            from prowler.providers.oraclecloud.services.identity.identity_password_policy_expires_within_365_days.identity_password_policy_expires_within_365_days import (
-                identity_password_policy_expires_within_365_days,
-            )
-
-            check = identity_password_policy_expires_within_365_days()
-            result = check.execute()
-
-            assert isinstance(result, list)
-
-            # If results exist, verify PASS findings
-            if len(result) > 0:
-                # Find PASS results
-                pass_results = [r for r in result if r.status == "PASS"]
-
-                if pass_results:
-                    # Detailed assertions on first PASS result
-                    assert pass_results[0].status == "PASS"
-                    assert pass_results[0].status_extended is not None
-                    assert len(pass_results[0].status_extended) > 0
-
-                    # Verify resource identification
-                    assert pass_results[0].resource_id is not None
-                    assert pass_results[0].resource_name is not None
-                    assert pass_results[0].region is not None
-                    assert pass_results[0].compartment_id is not None
-
-                    # Verify metadata
-                    assert pass_results[0].check_metadata.Provider == "oraclecloud"
-                    assert (
-                        pass_results[0].check_metadata.CheckID
-                        == "identity_password_policy_expires_within_365_days"
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=90,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
                     )
-                    assert pass_results[0].check_metadata.ServiceName == "identity"
-
-    def test_resource_non_compliant(self):
-        """identity_password_policy_expires_within_365_days: Resource fails the check (FAIL)"""
-        identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
-        identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock a non-compliant resource
-        resource = mock.MagicMock()
-        resource.id = "ocid1.resource.oc1.iad.bbbbbbbexample"
-        resource.name = "non-compliant-resource"
-        resource.region = OCI_REGION
-        resource.compartment_id = OCI_COMPARTMENT_ID
-        resource.lifecycle_state = "ACTIVE"
-        resource.tags = {"Environment": "Development"}
-
-        # Set attributes that make the resource non-compliant
-        resource.versioning = "Disabled"
-        resource.is_auto_rotation_enabled = False
-        resource.rotation_interval_in_days = None
-        resource.public_access_type = "ObjectRead"
-        resource.logging_enabled = False
-        resource.kms_key_id = None
-        resource.in_transit_encryption = "DISABLED"
-        resource.is_secure_boot_enabled = False
-        resource.legacy_endpoint_disabled = False
-        resource.is_legacy_imds_endpoint_disabled = False
-
-        # Mock client with non-compliant resource
-        identity_client.buckets = [resource]
-        identity_client.keys = [resource]
-        identity_client.volumes = [resource]
-        identity_client.boot_volumes = [resource]
-        identity_client.instances = [resource]
-        identity_client.file_systems = [resource]
-        identity_client.databases = [resource]
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.rules = []
-        identity_client.configuration = resource
-        identity_client.users = []
+                ]
+            )
+        ]
 
         with (
             mock.patch(
@@ -201,29 +99,139 @@ class Test_identity_password_policy_expires_within_365_days:
             check = identity_password_policy_expires_within_365_days()
             result = check.execute()
 
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "90 days" in result[0].status_extended
+            assert result[0].resource_id == POLICY_ID
 
-            # Verify FAIL findings exist
-            if len(result) > 0:
-                # Find FAIL results
-                fail_results = [r for r in result if r.status == "FAIL"]
-
-                if fail_results:
-                    # Detailed assertions on first FAIL result
-                    assert fail_results[0].status == "FAIL"
-                    assert fail_results[0].status_extended is not None
-                    assert len(fail_results[0].status_extended) > 0
-
-                    # Verify resource identification
-                    assert fail_results[0].resource_id is not None
-                    assert fail_results[0].resource_name is not None
-                    assert fail_results[0].region is not None
-                    assert fail_results[0].compartment_id is not None
-
-                    # Verify metadata
-                    assert fail_results[0].check_metadata.Provider == "oraclecloud"
-                    assert (
-                        fail_results[0].check_metadata.CheckID
-                        == "identity_password_policy_expires_within_365_days"
+    def test_policy_expires_over_365_days(self):
+        """Password expires after more than 365 days → FAIL."""
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=500,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
                     )
-                    assert fail_results[0].check_metadata.ServiceName == "identity"
+                ]
+            )
+        ]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_expires_within_365_days.identity_password_policy_expires_within_365_days.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_expires_within_365_days.identity_password_policy_expires_within_365_days import (
+                identity_password_policy_expires_within_365_days,
+            )
+
+            check = identity_password_policy_expires_within_365_days()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "500 days" in result[0].status_extended
+
+    def test_policy_no_expiration_configured(self):
+        """No password expiration configured → FAIL."""
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=None,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
+                    )
+                ]
+            )
+        ]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_expires_within_365_days.identity_password_policy_expires_within_365_days.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_expires_within_365_days.identity_password_policy_expires_within_365_days import (
+                identity_password_policy_expires_within_365_days,
+            )
+
+            check = identity_password_policy_expires_within_365_days()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "does not have password expiration" in result[0].status_extended
+
+    def test_system_managed_policies_excluded(self):
+        """System-managed policies should not appear in domain.password_policies.
+
+        This is a regression test: SimplePasswordPolicy and StandardPasswordPolicy
+        are filtered at the service layer, so checks never see them.
+        """
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        # Only user-configurable policy in the domain (system ones filtered by service)
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=90,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
+                    )
+                ]
+            )
+        ]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_expires_within_365_days.identity_password_policy_expires_within_365_days.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_expires_within_365_days.identity_password_policy_expires_within_365_days import (
+                identity_password_policy_expires_within_365_days,
+            )
+
+            check = identity_password_policy_expires_within_365_days()
+            result = check.execute()
+
+            # Only 1 finding for the custom policy, none for system-managed
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == POLICY_ID

tests/providers/oraclecloud/services/identity/identity_password_policy_minimum_length_14/identity_password_policy_minimum_length_14_test.py

@@ -1,5 +1,11 @@
+from datetime import datetime, timezone
 from unittest import mock
 
+from prowler.providers.oraclecloud.services.identity.identity_service import (
+    DomainPasswordPolicy,
+    IdentityDomain,
+    PasswordPolicy,
+)
 from tests.providers.oraclecloud.oci_fixtures import (
     OCI_COMPARTMENT_ID,
     OCI_REGION,
@@ -7,36 +13,36 @@ from tests.providers.oraclecloud.oci_fixtures import (
     set_mocked_oraclecloud_provider,
 )
 
+DOMAIN_ID = "ocid1.domain.oc1..aaaaaaaexample"
+DOMAIN_NAME = "Default"
+DOMAIN_URL = "https://idcs-example.identity.oraclecloud.com"
+POLICY_ID = "ocid1.passwordpolicy.oc1..aaaaaaaexample"
+POLICY_NAME = "CustomPasswordPolicy"
+
+
+def _make_domain(password_policies=None):
+    return IdentityDomain(
+        id=DOMAIN_ID,
+        display_name=DOMAIN_NAME,
+        description="Default identity domain",
+        url=DOMAIN_URL,
+        home_region=OCI_REGION,
+        compartment_id=OCI_COMPARTMENT_ID,
+        lifecycle_state="ACTIVE",
+        time_created=datetime.now(timezone.utc),
+        region=OCI_REGION,
+        password_policies=password_policies or [],
+    )
+
 
 class Test_identity_password_policy_minimum_length_14:
-    def test_no_resources(self):
-        """identity_password_policy_minimum_length_14: No resources to check"""
+    def test_no_domains_no_legacy_policy(self):
+        """No domains and no legacy policy → FAIL."""
         identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock empty collections
-        identity_client.rules = []
-        identity_client.topics = []
-        identity_client.subscriptions = []
-        identity_client.users = []
-        identity_client.groups = []
-        identity_client.policies = []
-        identity_client.compartments = []
-        identity_client.instances = []
-        identity_client.volumes = []
-        identity_client.boot_volumes = []
-        identity_client.buckets = []
-        identity_client.keys = []
-        identity_client.file_systems = []
-        identity_client.databases = []
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.subnets = []
-        identity_client.vcns = []
-        identity_client.configuration = None
-        identity_client.active_non_root_compartments = []
+        identity_client.domains = []
         identity_client.password_policy = None
+        identity_client.provider.identity.region = OCI_REGION
 
         with (
             mock.patch(
@@ -55,49 +61,30 @@ class Test_identity_password_policy_minimum_length_14:
             check = identity_password_policy_minimum_length_14()
             result = check.execute()
 
-            # Verify result is a list (empty or with findings)
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "No password policy" in result[0].status_extended
 
-    def test_resource_compliant(self):
-        """identity_password_policy_minimum_length_14: Resource passes the check (PASS)"""
+    def test_domain_policy_min_length_14(self):
+        """Domain password policy with min_length >= 14 → PASS."""
         identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock a compliant resource
-        resource = mock.MagicMock()
-        resource.id = "ocid1.resource.oc1.iad.aaaaaaaexample"
-        resource.name = "compliant-resource"
-        resource.region = OCI_REGION
-        resource.compartment_id = OCI_COMPARTMENT_ID
-        resource.lifecycle_state = "ACTIVE"
-        resource.tags = {"Environment": "Production"}
-
-        # Set attributes that make the resource compliant
-        resource.versioning = "Enabled"
-        resource.is_auto_rotation_enabled = True
-        resource.rotation_interval_in_days = 90
-        resource.public_access_type = "NoPublicAccess"
-        resource.logging_enabled = True
-        resource.kms_key_id = "ocid1.key.oc1.iad.aaaaaaaexample"
-        resource.in_transit_encryption = "ENABLED"
-        resource.is_secure_boot_enabled = True
-        resource.legacy_endpoint_disabled = True
-        resource.is_legacy_imds_endpoint_disabled = True
-
-        # Mock client with compliant resource
-        identity_client.buckets = [resource]
-        identity_client.keys = [resource]
-        identity_client.volumes = [resource]
-        identity_client.boot_volumes = [resource]
-        identity_client.instances = [resource]
-        identity_client.file_systems = [resource]
-        identity_client.databases = [resource]
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.rules = []
-        identity_client.configuration = resource
-        identity_client.users = []
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=90,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
+                    )
+                ]
+            )
+        ]
 
         with (
             mock.patch(
@@ -116,73 +103,31 @@ class Test_identity_password_policy_minimum_length_14:
             check = identity_password_policy_minimum_length_14()
             result = check.execute()
 
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "14 characters" in result[0].status_extended
+            assert result[0].resource_id == POLICY_ID
 
-            # If results exist, verify PASS findings
-            if len(result) > 0:
-                # Find PASS results
-                pass_results = [r for r in result if r.status == "PASS"]
-
-                if pass_results:
-                    # Detailed assertions on first PASS result
-                    assert pass_results[0].status == "PASS"
-                    assert pass_results[0].status_extended is not None
-                    assert len(pass_results[0].status_extended) > 0
-
-                    # Verify resource identification
-                    assert pass_results[0].resource_id is not None
-                    assert pass_results[0].resource_name is not None
-                    assert pass_results[0].region is not None
-                    assert pass_results[0].compartment_id is not None
-
-                    # Verify metadata
-                    assert pass_results[0].check_metadata.Provider == "oraclecloud"
-                    assert (
-                        pass_results[0].check_metadata.CheckID
-                        == "identity_password_policy_minimum_length_14"
-                    )
-                    assert pass_results[0].check_metadata.ServiceName == "identity"
-
-    def test_resource_non_compliant(self):
-        """identity_password_policy_minimum_length_14: Resource fails the check (FAIL)"""
+    def test_domain_policy_min_length_too_short(self):
+        """Domain password policy with min_length < 14 → FAIL."""
         identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock a non-compliant resource
-        resource = mock.MagicMock()
-        resource.id = "ocid1.resource.oc1.iad.bbbbbbbexample"
-        resource.name = "non-compliant-resource"
-        resource.region = OCI_REGION
-        resource.compartment_id = OCI_COMPARTMENT_ID
-        resource.lifecycle_state = "ACTIVE"
-        resource.tags = {"Environment": "Development"}
-
-        # Set attributes that make the resource non-compliant
-        resource.versioning = "Disabled"
-        resource.is_auto_rotation_enabled = False
-        resource.rotation_interval_in_days = None
-        resource.public_access_type = "ObjectRead"
-        resource.logging_enabled = False
-        resource.kms_key_id = None
-        resource.in_transit_encryption = "DISABLED"
-        resource.is_secure_boot_enabled = False
-        resource.legacy_endpoint_disabled = False
-        resource.is_legacy_imds_endpoint_disabled = False
-
-        # Mock client with non-compliant resource
-        identity_client.buckets = [resource]
-        identity_client.keys = [resource]
-        identity_client.volumes = [resource]
-        identity_client.boot_volumes = [resource]
-        identity_client.instances = [resource]
-        identity_client.file_systems = [resource]
-        identity_client.databases = [resource]
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.rules = []
-        identity_client.configuration = resource
-        identity_client.users = []
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=8,
+                        password_expires_after=90,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
+                    )
+                ]
+            )
+        ]
 
         with (
             mock.patch(
@@ -201,29 +146,116 @@ class Test_identity_password_policy_minimum_length_14:
             check = identity_password_policy_minimum_length_14()
             result = check.execute()
 
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "8" in result[0].status_extended
 
-            # Verify FAIL findings exist
-            if len(result) > 0:
-                # Find FAIL results
-                fail_results = [r for r in result if r.status == "FAIL"]
+    def test_legacy_policy_compliant(self):
+        """Legacy password policy with min length >= 14 → PASS."""
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = []
+        identity_client.password_policy = PasswordPolicy(
+            is_lowercase_characters_required=True,
+            is_uppercase_characters_required=True,
+            is_numeric_characters_required=True,
+            is_special_characters_required=True,
+            is_username_containment_allowed=False,
+            minimum_password_length=14,
+        )
+        identity_client.provider.identity.region = OCI_REGION
 
-                if fail_results:
-                    # Detailed assertions on first FAIL result
-                    assert fail_results[0].status == "FAIL"
-                    assert fail_results[0].status_extended is not None
-                    assert len(fail_results[0].status_extended) > 0
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_minimum_length_14.identity_password_policy_minimum_length_14.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_minimum_length_14.identity_password_policy_minimum_length_14 import (
+                identity_password_policy_minimum_length_14,
+            )
 
-                    # Verify resource identification
-                    assert fail_results[0].resource_id is not None
-                    assert fail_results[0].resource_name is not None
-                    assert fail_results[0].region is not None
-                    assert fail_results[0].compartment_id is not None
+            check = identity_password_policy_minimum_length_14()
+            result = check.execute()
 
-                    # Verify metadata
-                    assert fail_results[0].check_metadata.Provider == "oraclecloud"
-                    assert (
-                        fail_results[0].check_metadata.CheckID
-                        == "identity_password_policy_minimum_length_14"
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "14 characters" in result[0].status_extended
+
+    def test_domain_no_policies(self):
+        """Domain with no password policies → FAIL."""
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = [_make_domain([])]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_minimum_length_14.identity_password_policy_minimum_length_14.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_minimum_length_14.identity_password_policy_minimum_length_14 import (
+                identity_password_policy_minimum_length_14,
+            )
+
+            check = identity_password_policy_minimum_length_14()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "no password policy configured" in result[0].status_extended
+
+    def test_system_managed_policies_excluded(self):
+        """System-managed policies should not appear in domain.password_policies.
+
+        This is a regression test: SimplePasswordPolicy and StandardPasswordPolicy
+        are filtered at the service layer, so checks never see them.
+        """
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=90,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
                     )
-                    assert fail_results[0].check_metadata.ServiceName == "identity"
+                ]
+            )
+        ]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_minimum_length_14.identity_password_policy_minimum_length_14.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_minimum_length_14.identity_password_policy_minimum_length_14 import (
+                identity_password_policy_minimum_length_14,
+            )
+
+            check = identity_password_policy_minimum_length_14()
+            result = check.execute()
+
+            # Only 1 finding for the custom policy, none for system-managed
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == POLICY_ID

tests/providers/oraclecloud/services/identity/identity_password_policy_prevents_reuse/identity_password_policy_prevents_reuse_test.py

@@ -1,5 +1,10 @@
+from datetime import datetime, timezone
 from unittest import mock
 
+from prowler.providers.oraclecloud.services.identity.identity_service import (
+    DomainPasswordPolicy,
+    IdentityDomain,
+)
 from tests.providers.oraclecloud.oci_fixtures import (
     OCI_COMPARTMENT_ID,
     OCI_REGION,
@@ -7,36 +12,34 @@ from tests.providers.oraclecloud.oci_fixtures import (
     set_mocked_oraclecloud_provider,
 )
 
+DOMAIN_ID = "ocid1.domain.oc1..aaaaaaaexample"
+DOMAIN_NAME = "Default"
+DOMAIN_URL = "https://idcs-example.identity.oraclecloud.com"
+POLICY_ID = "ocid1.passwordpolicy.oc1..aaaaaaaexample"
+POLICY_NAME = "CustomPasswordPolicy"
+
+
+def _make_domain(password_policies=None):
+    return IdentityDomain(
+        id=DOMAIN_ID,
+        display_name=DOMAIN_NAME,
+        description="Default identity domain",
+        url=DOMAIN_URL,
+        home_region=OCI_REGION,
+        compartment_id=OCI_COMPARTMENT_ID,
+        lifecycle_state="ACTIVE",
+        time_created=datetime.now(timezone.utc),
+        region=OCI_REGION,
+        password_policies=password_policies or [],
+    )
+
 
 class Test_identity_password_policy_prevents_reuse:
-    def test_no_resources(self):
-        """identity_password_policy_prevents_reuse: No resources to check"""
+    def test_no_domains(self):
+        """No Identity Domains → MANUAL finding."""
         identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock empty collections
-        identity_client.rules = []
-        identity_client.topics = []
-        identity_client.subscriptions = []
-        identity_client.users = []
-        identity_client.groups = []
-        identity_client.policies = []
-        identity_client.compartments = []
-        identity_client.instances = []
-        identity_client.volumes = []
-        identity_client.boot_volumes = []
-        identity_client.buckets = []
-        identity_client.keys = []
-        identity_client.file_systems = []
-        identity_client.databases = []
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.subnets = []
-        identity_client.vcns = []
-        identity_client.configuration = None
-        identity_client.active_non_root_compartments = []
-        identity_client.password_policy = None
+        identity_client.domains = []
 
         with (
             mock.patch(
@@ -55,134 +58,29 @@ class Test_identity_password_policy_prevents_reuse:
             check = identity_password_policy_prevents_reuse()
             result = check.execute()
 
-            # Verify result is a list (empty or with findings)
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "MANUAL"
 
-    def test_resource_compliant(self):
-        """identity_password_policy_prevents_reuse: Resource passes the check (PASS)"""
+    def test_policy_prevents_reuse_24(self):
+        """Password history >= 24 → PASS."""
         identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock a compliant resource
-        resource = mock.MagicMock()
-        resource.id = "ocid1.resource.oc1.iad.aaaaaaaexample"
-        resource.name = "compliant-resource"
-        resource.region = OCI_REGION
-        resource.compartment_id = OCI_COMPARTMENT_ID
-        resource.lifecycle_state = "ACTIVE"
-        resource.tags = {"Environment": "Production"}
-
-        # Set attributes that make the resource compliant
-        resource.versioning = "Enabled"
-        resource.is_auto_rotation_enabled = True
-        resource.rotation_interval_in_days = 90
-        resource.public_access_type = "NoPublicAccess"
-        resource.logging_enabled = True
-        resource.kms_key_id = "ocid1.key.oc1.iad.aaaaaaaexample"
-        resource.in_transit_encryption = "ENABLED"
-        resource.is_secure_boot_enabled = True
-        resource.legacy_endpoint_disabled = True
-        resource.is_legacy_imds_endpoint_disabled = True
-
-        # Mock client with compliant resource
-        identity_client.buckets = [resource]
-        identity_client.keys = [resource]
-        identity_client.volumes = [resource]
-        identity_client.boot_volumes = [resource]
-        identity_client.instances = [resource]
-        identity_client.file_systems = [resource]
-        identity_client.databases = [resource]
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.rules = []
-        identity_client.configuration = resource
-        identity_client.users = []
-
-        with (
-            mock.patch(
-                "prowler.providers.common.provider.Provider.get_global_provider",
-                return_value=set_mocked_oraclecloud_provider(),
-            ),
-            mock.patch(
-                "prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse.identity_client",
-                new=identity_client,
-            ),
-        ):
-            from prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse import (
-                identity_password_policy_prevents_reuse,
-            )
-
-            check = identity_password_policy_prevents_reuse()
-            result = check.execute()
-
-            assert isinstance(result, list)
-
-            # If results exist, verify PASS findings
-            if len(result) > 0:
-                # Find PASS results
-                pass_results = [r for r in result if r.status == "PASS"]
-
-                if pass_results:
-                    # Detailed assertions on first PASS result
-                    assert pass_results[0].status == "PASS"
-                    assert pass_results[0].status_extended is not None
-                    assert len(pass_results[0].status_extended) > 0
-
-                    # Verify resource identification
-                    assert pass_results[0].resource_id is not None
-                    assert pass_results[0].resource_name is not None
-                    assert pass_results[0].region is not None
-                    assert pass_results[0].compartment_id is not None
-
-                    # Verify metadata
-                    assert pass_results[0].check_metadata.Provider == "oraclecloud"
-                    assert (
-                        pass_results[0].check_metadata.CheckID
-                        == "identity_password_policy_prevents_reuse"
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=90,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
                     )
-                    assert pass_results[0].check_metadata.ServiceName == "identity"
-
-    def test_resource_non_compliant(self):
-        """identity_password_policy_prevents_reuse: Resource fails the check (FAIL)"""
-        identity_client = mock.MagicMock()
-        identity_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
-        identity_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock a non-compliant resource
-        resource = mock.MagicMock()
-        resource.id = "ocid1.resource.oc1.iad.bbbbbbbexample"
-        resource.name = "non-compliant-resource"
-        resource.region = OCI_REGION
-        resource.compartment_id = OCI_COMPARTMENT_ID
-        resource.lifecycle_state = "ACTIVE"
-        resource.tags = {"Environment": "Development"}
-
-        # Set attributes that make the resource non-compliant
-        resource.versioning = "Disabled"
-        resource.is_auto_rotation_enabled = False
-        resource.rotation_interval_in_days = None
-        resource.public_access_type = "ObjectRead"
-        resource.logging_enabled = False
-        resource.kms_key_id = None
-        resource.in_transit_encryption = "DISABLED"
-        resource.is_secure_boot_enabled = False
-        resource.legacy_endpoint_disabled = False
-        resource.is_legacy_imds_endpoint_disabled = False
-
-        # Mock client with non-compliant resource
-        identity_client.buckets = [resource]
-        identity_client.keys = [resource]
-        identity_client.volumes = [resource]
-        identity_client.boot_volumes = [resource]
-        identity_client.instances = [resource]
-        identity_client.file_systems = [resource]
-        identity_client.databases = [resource]
-        identity_client.security_lists = []
-        identity_client.security_groups = []
-        identity_client.rules = []
-        identity_client.configuration = resource
-        identity_client.users = []
+                ]
+            )
+        ]
 
         with (
             mock.patch(
@@ -201,29 +99,165 @@ class Test_identity_password_policy_prevents_reuse:
             check = identity_password_policy_prevents_reuse()
             result = check.execute()
 
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "24 passwords" in result[0].status_extended
+            assert result[0].resource_id == POLICY_ID
 
-            # Verify FAIL findings exist
-            if len(result) > 0:
-                # Find FAIL results
-                fail_results = [r for r in result if r.status == "FAIL"]
-
-                if fail_results:
-                    # Detailed assertions on first FAIL result
-                    assert fail_results[0].status == "FAIL"
-                    assert fail_results[0].status_extended is not None
-                    assert len(fail_results[0].status_extended) > 0
-
-                    # Verify resource identification
-                    assert fail_results[0].resource_id is not None
-                    assert fail_results[0].resource_name is not None
-                    assert fail_results[0].region is not None
-                    assert fail_results[0].compartment_id is not None
-
-                    # Verify metadata
-                    assert fail_results[0].check_metadata.Provider == "oraclecloud"
-                    assert (
-                        fail_results[0].check_metadata.CheckID
-                        == "identity_password_policy_prevents_reuse"
+    def test_policy_insufficient_history(self):
+        """Password history < 24 → FAIL."""
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=90,
+                        num_passwords_in_history=5,
+                        password_expire_warning=7,
+                        min_password_age=1,
                     )
-                    assert fail_results[0].check_metadata.ServiceName == "identity"
+                ]
+            )
+        ]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse import (
+                identity_password_policy_prevents_reuse,
+            )
+
+            check = identity_password_policy_prevents_reuse()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "5 passwords" in result[0].status_extended
+
+    def test_policy_no_history_configured(self):
+        """No password history configured → FAIL."""
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=90,
+                        num_passwords_in_history=None,
+                        password_expire_warning=7,
+                        min_password_age=1,
+                    )
+                ]
+            )
+        ]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse import (
+                identity_password_policy_prevents_reuse,
+            )
+
+            check = identity_password_policy_prevents_reuse()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "does not have password history" in result[0].status_extended
+
+    def test_domain_no_policies(self):
+        """Domain with no password policies → FAIL."""
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = [_make_domain([])]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse import (
+                identity_password_policy_prevents_reuse,
+            )
+
+            check = identity_password_policy_prevents_reuse()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "no password policy configured" in result[0].status_extended
+
+    def test_system_managed_policies_excluded(self):
+        """System-managed policies should not appear in domain.password_policies.
+
+        This is a regression test: SimplePasswordPolicy and StandardPasswordPolicy
+        are filtered at the service layer, so checks never see them.
+        """
+        identity_client = mock.MagicMock()
+        identity_client.audited_tenancy = OCI_TENANCY_ID
+        identity_client.domains = [
+            _make_domain(
+                [
+                    DomainPasswordPolicy(
+                        id=POLICY_ID,
+                        name=POLICY_NAME,
+                        description="Custom policy",
+                        min_length=14,
+                        password_expires_after=90,
+                        num_passwords_in_history=24,
+                        password_expire_warning=7,
+                        min_password_age=1,
+                    )
+                ]
+            )
+        ]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse.identity_client",
+                new=identity_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.identity.identity_password_policy_prevents_reuse.identity_password_policy_prevents_reuse import (
+                identity_password_policy_prevents_reuse,
+            )
+
+            check = identity_password_policy_prevents_reuse()
+            result = check.execute()
+
+            # Only 1 finding for the custom policy, none for system-managed
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].resource_id == POLICY_ID

tests/providers/oraclecloud/services/kms/kms_key_rotation_enabled/oraclecloud_kms_key_rotation_enabled_test.py

@@ -1,5 +1,7 @@
+from datetime import datetime, timedelta, timezone
 from unittest import mock
 
+from prowler.providers.oraclecloud.services.kms.kms_service import Key
 from tests.providers.oraclecloud.oci_fixtures import (
     OCI_COMPARTMENT_ID,
     OCI_REGION,
@@ -7,36 +9,17 @@ from tests.providers.oraclecloud.oci_fixtures import (
     set_mocked_oraclecloud_provider,
 )
 
+KEY_ID = "ocid1.key.oc1.iad.aaaaaaaexample"
+KEY_NAME = "test-key"
+
 
 class Test_kms_key_rotation_enabled:
-    def test_no_resources(self):
-        """kms_key_rotation_enabled: No resources to check"""
+    def test_no_keys(self):
+        """No keys → empty findings."""
         kms_client = mock.MagicMock()
         kms_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         kms_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock empty collections
-        kms_client.rules = []
-        kms_client.topics = []
-        kms_client.subscriptions = []
-        kms_client.users = []
-        kms_client.groups = []
-        kms_client.policies = []
-        kms_client.compartments = []
-        kms_client.instances = []
-        kms_client.volumes = []
-        kms_client.boot_volumes = []
-        kms_client.buckets = []
         kms_client.keys = []
-        kms_client.file_systems = []
-        kms_client.databases = []
-        kms_client.security_lists = []
-        kms_client.security_groups = []
-        kms_client.subnets = []
-        kms_client.vcns = []
-        kms_client.configuration = None
-        kms_client.active_non_root_compartments = []
-        kms_client.password_policy = None
 
         with (
             mock.patch(
@@ -55,49 +38,24 @@ class Test_kms_key_rotation_enabled:
             check = kms_key_rotation_enabled()
             result = check.execute()
 
-            # Verify result is a list (empty or with findings)
-            assert isinstance(result, list)
+            assert result == []
 
-    def test_resource_compliant(self):
-        """kms_key_rotation_enabled: Resource passes the check (PASS)"""
+    def test_key_with_auto_rotation_enabled(self):
+        """Key with auto-rotation enabled → PASS."""
         kms_client = mock.MagicMock()
         kms_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         kms_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock a compliant resource
-        resource = mock.MagicMock()
-        resource.id = "ocid1.resource.oc1.iad.aaaaaaaexample"
-        resource.name = "compliant-resource"
-        resource.region = OCI_REGION
-        resource.compartment_id = OCI_COMPARTMENT_ID
-        resource.lifecycle_state = "ACTIVE"
-        resource.tags = {"Environment": "Production"}
-
-        # Set attributes that make the resource compliant
-        resource.versioning = "Enabled"
-        resource.is_auto_rotation_enabled = True
-        resource.rotation_interval_in_days = 90
-        resource.public_access_type = "NoPublicAccess"
-        resource.logging_enabled = True
-        resource.kms_key_id = "ocid1.key.oc1.iad.aaaaaaaexample"
-        resource.in_transit_encryption = "ENABLED"
-        resource.is_secure_boot_enabled = True
-        resource.legacy_endpoint_disabled = True
-        resource.is_legacy_imds_endpoint_disabled = True
-
-        # Mock client with compliant resource
-        kms_client.buckets = [resource]
-        kms_client.keys = [resource]
-        kms_client.volumes = [resource]
-        kms_client.boot_volumes = [resource]
-        kms_client.instances = [resource]
-        kms_client.file_systems = [resource]
-        kms_client.databases = [resource]
-        kms_client.security_lists = []
-        kms_client.security_groups = []
-        kms_client.rules = []
-        kms_client.configuration = resource
-        kms_client.users = []
+        kms_client.keys = [
+            Key(
+                id=KEY_ID,
+                name=KEY_NAME,
+                compartment_id=OCI_COMPARTMENT_ID,
+                region=OCI_REGION,
+                lifecycle_state="ENABLED",
+                is_auto_rotation_enabled=True,
+                rotation_interval_in_days=90,
+            )
+        ]
 
         with (
             mock.patch(
@@ -116,73 +74,32 @@ class Test_kms_key_rotation_enabled:
             check = kms_key_rotation_enabled()
             result = check.execute()
 
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "auto-rotation enabled" in result[0].status_extended
+            assert result[0].resource_id == KEY_ID
+            assert result[0].resource_name == KEY_NAME
+            assert result[0].region == OCI_REGION
+            assert result[0].compartment_id == OCI_COMPARTMENT_ID
 
-            # If results exist, verify PASS findings
-            if len(result) > 0:
-                # Find PASS results
-                pass_results = [r for r in result if r.status == "PASS"]
-
-                if pass_results:
-                    # Detailed assertions on first PASS result
-                    assert pass_results[0].status == "PASS"
-                    assert pass_results[0].status_extended is not None
-                    assert len(pass_results[0].status_extended) > 0
-
-                    # Verify resource identification
-                    assert pass_results[0].resource_id is not None
-                    assert pass_results[0].resource_name is not None
-                    assert pass_results[0].region is not None
-                    assert pass_results[0].compartment_id is not None
-
-                    # Verify metadata
-                    assert pass_results[0].check_metadata.Provider == "oraclecloud"
-                    assert (
-                        pass_results[0].check_metadata.CheckID
-                        == "kms_key_rotation_enabled"
-                    )
-                    assert pass_results[0].check_metadata.ServiceName == "kms"
-
-    def test_resource_non_compliant(self):
-        """kms_key_rotation_enabled: Resource fails the check (FAIL)"""
+    def test_key_manually_rotated_within_365_days(self):
+        """Key manually rotated within last 365 days (no auto-rotation) → PASS."""
         kms_client = mock.MagicMock()
         kms_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
         kms_client.audited_tenancy = OCI_TENANCY_ID
-
-        # Mock a non-compliant resource
-        resource = mock.MagicMock()
-        resource.id = "ocid1.resource.oc1.iad.bbbbbbbexample"
-        resource.name = "non-compliant-resource"
-        resource.region = OCI_REGION
-        resource.compartment_id = OCI_COMPARTMENT_ID
-        resource.lifecycle_state = "ACTIVE"
-        resource.tags = {"Environment": "Development"}
-
-        # Set attributes that make the resource non-compliant
-        resource.versioning = "Disabled"
-        resource.is_auto_rotation_enabled = False
-        resource.rotation_interval_in_days = None
-        resource.public_access_type = "ObjectRead"
-        resource.logging_enabled = False
-        resource.kms_key_id = None
-        resource.in_transit_encryption = "DISABLED"
-        resource.is_secure_boot_enabled = False
-        resource.legacy_endpoint_disabled = False
-        resource.is_legacy_imds_endpoint_disabled = False
-
-        # Mock client with non-compliant resource
-        kms_client.buckets = [resource]
-        kms_client.keys = [resource]
-        kms_client.volumes = [resource]
-        kms_client.boot_volumes = [resource]
-        kms_client.instances = [resource]
-        kms_client.file_systems = [resource]
-        kms_client.databases = [resource]
-        kms_client.security_lists = []
-        kms_client.security_groups = []
-        kms_client.rules = []
-        kms_client.configuration = resource
-        kms_client.users = []
+        kms_client.keys = [
+            Key(
+                id=KEY_ID,
+                name=KEY_NAME,
+                compartment_id=OCI_COMPARTMENT_ID,
+                region=OCI_REGION,
+                lifecycle_state="ENABLED",
+                is_auto_rotation_enabled=False,
+                rotation_interval_in_days=None,
+                current_key_version_time_created=datetime.now(timezone.utc)
+                - timedelta(days=100),
+            )
+        ]
 
         with (
             mock.patch(
@@ -201,29 +118,88 @@ class Test_kms_key_rotation_enabled:
             check = kms_key_rotation_enabled()
             result = check.execute()
 
-            assert isinstance(result, list)
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "manually rotated" in result[0].status_extended
+            assert result[0].resource_id == KEY_ID
 
-            # Verify FAIL findings exist
-            if len(result) > 0:
-                # Find FAIL results
-                fail_results = [r for r in result if r.status == "FAIL"]
+    def test_key_manually_rotated_over_365_days_ago(self):
+        """Key manually rotated more than 365 days ago (no auto-rotation) → FAIL."""
+        kms_client = mock.MagicMock()
+        kms_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
+        kms_client.audited_tenancy = OCI_TENANCY_ID
+        kms_client.keys = [
+            Key(
+                id=KEY_ID,
+                name=KEY_NAME,
+                compartment_id=OCI_COMPARTMENT_ID,
+                region=OCI_REGION,
+                lifecycle_state="ENABLED",
+                is_auto_rotation_enabled=False,
+                rotation_interval_in_days=None,
+                current_key_version_time_created=datetime.now(timezone.utc)
+                - timedelta(days=400),
+            )
+        ]
 
-                if fail_results:
-                    # Detailed assertions on first FAIL result
-                    assert fail_results[0].status == "FAIL"
-                    assert fail_results[0].status_extended is not None
-                    assert len(fail_results[0].status_extended) > 0
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled.kms_client",
+                new=kms_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled import (
+                kms_key_rotation_enabled,
+            )
 
-                    # Verify resource identification
-                    assert fail_results[0].resource_id is not None
-                    assert fail_results[0].resource_name is not None
-                    assert fail_results[0].region is not None
-                    assert fail_results[0].compartment_id is not None
+            check = kms_key_rotation_enabled()
+            result = check.execute()
 
-                    # Verify metadata
-                    assert fail_results[0].check_metadata.Provider == "oraclecloud"
-                    assert (
-                        fail_results[0].check_metadata.CheckID
-                        == "kms_key_rotation_enabled"
-                    )
-                    assert fail_results[0].check_metadata.ServiceName == "kms"
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "not been rotated" in result[0].status_extended
+            assert result[0].resource_id == KEY_ID
+
+    def test_key_no_rotation_at_all(self):
+        """Key with no auto-rotation and no version info → FAIL."""
+        kms_client = mock.MagicMock()
+        kms_client.audited_compartments = {OCI_COMPARTMENT_ID: mock.MagicMock()}
+        kms_client.audited_tenancy = OCI_TENANCY_ID
+        kms_client.keys = [
+            Key(
+                id=KEY_ID,
+                name=KEY_NAME,
+                compartment_id=OCI_COMPARTMENT_ID,
+                region=OCI_REGION,
+                lifecycle_state="ENABLED",
+                is_auto_rotation_enabled=False,
+                rotation_interval_in_days=None,
+                current_key_version_time_created=None,
+            )
+        ]
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_oraclecloud_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.oraclecloud.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled.kms_client",
+                new=kms_client,
+            ),
+        ):
+            from prowler.providers.oraclecloud.services.kms.kms_key_rotation_enabled.kms_key_rotation_enabled import (
+                kms_key_rotation_enabled,
+            )
+
+            check = kms_key_rotation_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert "not been rotated" in result[0].status_extended
+            assert result[0].resource_id == KEY_ID