fix: remove unnecessary changes

2026-07-24 04:51:51 +00:00 · 2025-06-04 13:21:25 +02:00
parent 00e491415f
commit 9d4094e19e
4 changed files with 79 additions and 181 deletions
@@ -1,62 +1,42 @@
-from typing import Dict, Optional
-
 from prowler.lib.logger import logger
-from prowler.providers.aws.lib.fix.fixer import AWSFixer, FixerMetadata
+from prowler.providers.aws.services.accessanalyzer.accessanalyzer_client import (
+    accessanalyzer_client,
+)


-class AccessAnalyzerEnabledFixer(AWSFixer):
-    """Fixer for AccessAnalyzer enabled check"""
-
-    def __init__(
-        self, credentials: Optional[Dict] = None, session_config: Optional[Dict] = None
-    ):
-        super().__init__(credentials, session_config)
-        self.service = "accessanalyzer"
-        self.regional_clients = ["accessanalyzer"]
-        self.iam_policy_required = {
-            "Version": "2012-10-17",
-            "Statement": [
-                {
-                    "Effect": "Allow",
-                    "Action": "access-analyzer:CreateAnalyzer",
-                    "Resource": "*",
-                }
-            ],
-        }
-
-    def _get_metadata(self) -> FixerMetadata:
-        return FixerMetadata(
-            description="Enable Access Analyzer in a region",
-            cost_impact=True,
-            cost_description="Enabling Access Analyzer may incur AWS charges",
+def fixer(region):
+    """
+    Enable Access Analyzer in a region. Requires the access-analyzer:CreateAnalyzer permission.
+    Permissions:
+    {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Action": "access-analyzer:CreateAnalyzer",
+                "Resource": "*"
+            }
+        ]
+    }
+    Args:
+        region (str): AWS region
+    Returns:
+        bool: True if Access Analyzer is enabled, False otherwise
+    """
+    try:
+        regional_client = accessanalyzer_client.regional_clients[region]
+        regional_client.create_analyzer(
+            analyzerName=accessanalyzer_client.fixer_config.get(
+                "accessanalyzer_enabled", {}
+            ).get("AnalyzerName", "DefaultAnalyzer"),
+            type=accessanalyzer_client.fixer_config.get(
+                "accessanalyzer_enabled", {}
+            ).get("AnalyzerType", "ACCOUNT_UNUSED_ACCESS"),
        )
-
-    def fix(self, finding: Optional[Dict] = None, **kwargs) -> bool:
-        """
-        Enable Access Analyzer in a region
-        Args:
-            finding: Finding dictionary (optional)
-            **kwargs: Additional arguments (region is required if finding is not provided)
-        Returns:
-            bool: True if Access Analyzer is enabled, False otherwise
-        """
-        try:
-            region = kwargs.get("region") or (finding and finding.get("Region"))
-            if not region:
-                raise ValueError("Region is required")
-
-            regional_client = self.client.regional_clients[region]
-            regional_client.create_analyzer(
-                analyzerName=self.client.fixer_config.get(
-                    "accessanalyzer_enabled", {}
-                ).get("AnalyzerName", "DefaultAnalyzer"),
-                type=self.client.fixer_config.get("accessanalyzer_enabled", {}).get(
-                    "AnalyzerType", "ACCOUNT_UNUSED_ACCESS"
-                ),
-            )
-            return True
-        except Exception as error:
-            logger.error(
-                f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-            )
-            return False
+    except Exception as error:
+        logger.error(
+            f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+        )
+        return False
+    else:
+        return True
@@ -1,71 +1,40 @@
-from typing import Dict, Optional
-
 from prowler.lib.logger import logger
-from prowler.providers.aws.lib.fix.fixer import AWSFixer, FixerMetadata
 from prowler.providers.aws.services.rds.rds_client import rds_client


-class RdsInstanceNoPublicAccessFixer(AWSFixer):
+def fixer(resource_id: str, region: str) -> bool:
    """
-    Fixer for RDS instances with public access
+    Modify the attributes of an RDS instance to disable public accessibility.
+    Specifically, this fixer sets the 'PubliclyAccessible' attribute to False
+    to prevent the RDS instance from being publicly accessible. Requires the rds:ModifyDBInstance permission.
+    Permissions:
+    {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Action": "rds:ModifyDBInstance",
+                "Resource": "*"
+            }
+        ]
+    }
+    Args:
+        resource_id (str): The DB instance identifier.
+        region (str): AWS region where the DB instance exists.
+    Returns:
+        bool: True if the operation is successful (public access is disabled), False otherwise.
    """
-
-    def __init__(
-        self, credentials: Optional[Dict] = None, session_config: Optional[Dict] = None
-    ):
-        super().__init__(credentials, session_config)
-        self.service = "rds"
-        self.iam_policy_required = {
-            "Version": "2012-10-17",
-            "Statement": [
-                {"Effect": "Allow", "Action": "rds:ModifyDBInstance", "Resource": "*"}
-            ],
-        }
-
-    def _get_metadata(self) -> FixerMetadata:
-        return FixerMetadata(
-            description="Disable public accessibility for RDS instance",
-            cost_impact=False,
-            cost_description=None,
+    try:
+        regional_client = rds_client.regional_clients[region]
+        regional_client.modify_db_instance(
+            DBInstanceIdentifier=resource_id,
+            PubliclyAccessible=False,
+            ApplyImmediately=True,
        )
-
-    def fix(self, finding: Optional[Dict] = None, **kwargs) -> bool:
-        """
-        Modify the attributes of an RDS instance to disable public accessibility.
-        Specifically, this fixer sets the 'PubliclyAccessible' attribute to False
-        to prevent the RDS instance from being publicly accessible.
-
-        Args:
-            finding (Dict): Finding to fix
-            **kwargs: Additional arguments (region and resource_id are required if finding is not provided)
-
-        Returns:
-            bool: True if the operation is successful (public access is disabled), False otherwise
-        """
-        try:
-            # Get region and resource_id either from finding or kwargs
-            if finding:
-                region = finding.get("Region") or finding.get("region")
-                resource_id = finding.get("ResourceId") or finding.get("resource_id")
-            else:
-                region = kwargs.get("region")
-                resource_id = kwargs.get("resource_id")
-
-            if not region or not resource_id:
-                raise ValueError("Region and resource_id are required")
-
-            # Get the client for this region
-            regional_client = rds_client.regional_clients[region]
-
-            # Modify the DB instance
-            regional_client.modify_db_instance(
-                DBInstanceIdentifier=resource_id,
-                PubliclyAccessible=False,
-                ApplyImmediately=True,
-            )
-            return True
-        except Exception as error:
-            logger.error(
-                f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-            )
-            return False
+    except Exception as error:
+        logger.error(
+            f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+        )
+        return False
+    else:
+        return True
@@ -32,13 +32,10 @@ class Test_rds_instance_no_public_access_fixer:
            ):
                # Test Fixer
                from prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access_fixer import (
-                    RdsInstanceNoPublicAccessFixer,
+                    fixer,
                )

-                fixer = RdsInstanceNoPublicAccessFixer()
-                assert fixer.fix(
-                    None, region=AWS_REGION_US_EAST_1, resource_id="db-primary-1"
-                )
+                assert fixer("db-primary-1", AWS_REGION_US_EAST_1)

    @mock_aws
    def test_rds_public(self):
@@ -64,24 +61,13 @@ class Test_rds_instance_no_public_access_fixer:
                "prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access_fixer.rds_client",
                new=RDS(aws_provider),
            ):
+
                # Test Fixer
                from prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access_fixer import (
-                    RdsInstanceNoPublicAccessFixer,
+                    fixer,
                )

-                fixer = RdsInstanceNoPublicAccessFixer()
-                # Test with finding
-                mock_finding = {
-                    "Region": AWS_REGION_US_EAST_1,
-                    "ResourceId": "db-primary-1",
-                    "Status": "FAIL",
-                }
-                assert fixer.fix(finding=mock_finding)
-
-                # Test with kwargs
-                assert fixer.fix(
-                    None, region=AWS_REGION_US_EAST_1, resource_id="db-primary-1"
-                )
+                assert fixer("db-primary-1", AWS_REGION_US_EAST_1)

    @mock_aws
    def test_rds_cluster_public_snapshot_error(self):
@@ -107,47 +93,10 @@ class Test_rds_instance_no_public_access_fixer:
                "prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access_fixer.rds_client",
                new=RDS(aws_provider),
            ):
+
                # Test Fixer
                from prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access_fixer import (
-                    RdsInstanceNoPublicAccessFixer,
+                    fixer,
                )

-                fixer = RdsInstanceNoPublicAccessFixer()
-                assert not fixer.fix(
-                    None, region=AWS_REGION_US_EAST_1, resource_id="db-primary-2"
-                )
-
-    @mock_aws
-    def test_rds_error_handling(self):
-        from prowler.providers.aws.services.rds.rds_service import RDS
-
-        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
-
-        with mock.patch(
-            "prowler.providers.common.provider.Provider.get_global_provider",
-            return_value=aws_provider,
-        ):
-            with mock.patch(
-                "prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access_fixer.rds_client",
-                new=RDS(aws_provider),
-            ):
-                # Test Fixer
-                from prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access_fixer import (
-                    RdsInstanceNoPublicAccessFixer,
-                )
-
-                fixer = RdsInstanceNoPublicAccessFixer()
-
-                # Test with non-existent instance
-                assert not fixer.fix(
-                    None, region=AWS_REGION_US_EAST_1, resource_id="db-primary-2"
-                )
-
-                # Test with missing parameters
-                assert not fixer.fix(None)
-                assert not fixer.fix(None, region=AWS_REGION_US_EAST_1)
-                assert not fixer.fix(None, resource_id="db-primary-1")
-
-                # Test with invalid finding format
-                invalid_finding = {"wrong_key": "wrong_value"}
-                assert not fixer.fix(finding=invalid_finding)
+                assert not fixer("db-primary-2", AWS_REGION_US_EAST_1)