From 18a6f29593553584dd7c068b656531e04bebced5 Mon Sep 17 00:00:00 2001 From: Daniel Barranquero Date: Tue, 3 Jun 2025 17:40:05 +0200 Subject: [PATCH] feat(gcp): add first version of gcp fixers --- prowler/providers/gcp/lib/fix/__init__.py | 0 prowler/providers/gcp/lib/fix/fixer.py | 97 +++++++++++++++++++ .../compute_project_os_login_enabled_fixer.py | 63 ++++++++++++ 3 files changed, 160 insertions(+) create mode 100644 prowler/providers/gcp/lib/fix/__init__.py create mode 100644 prowler/providers/gcp/lib/fix/fixer.py create mode 100644 prowler/providers/gcp/services/compute/compute_project_os_login_enabled/compute_project_os_login_enabled_fixer.py diff --git a/prowler/providers/gcp/lib/fix/__init__.py b/prowler/providers/gcp/lib/fix/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/gcp/lib/fix/fixer.py b/prowler/providers/gcp/lib/fix/fixer.py new file mode 100644 index 0000000000..1c1ef54787 --- /dev/null +++ b/prowler/providers/gcp/lib/fix/fixer.py @@ -0,0 +1,97 @@ +from typing import Dict, Optional + +from prowler.lib.check.models import Check_Report_GCP +from prowler.lib.fix.fixer import Fixer +from prowler.lib.logger import logger +from prowler.providers.gcp.gcp_provider import GcpProvider + + +class GCPFixer(Fixer): + """GCP specific fixer implementation""" + + def __init__( + self, + description: str, + cost_impact: bool = False, + cost_description: Optional[str] = None, + service: str = "", + iam_policy_required: Optional[Dict] = None, + ): + """ + Initialize GCP fixer with metadata. + + Args: + description (str): Description of the fixer + cost_impact (bool): Whether the fixer has a cost impact + cost_description (Optional[str]): Description of the cost impact + service (str): GCP service name + iam_policy_required (Optional[Dict]): Required IAM policy for the fixer + """ + super().__init__(description, cost_impact, cost_description) + self.service = service + self.iam_policy_required = iam_policy_required or {} + self._provider = None + + @property + def provider(self) -> GcpProvider: + """Get the GCP provider instance""" + if not self._provider: + self._provider = GcpProvider() + return self._provider + + def _get_fixer_info(self) -> Dict: + """Get fixer metadata""" + info = super()._get_fixer_info() + info["service"] = self.service + info["iam_policy_required"] = self.iam_policy_required + info["provider"] = "gcp" + return info + + def fix(self, finding: Optional[Check_Report_GCP] = None, **kwargs) -> bool: + """ + GCP specific method to execute the fixer. + This method handles the printing of fixing status messages. + + Args: + finding (Optional[Check_Report_GCP]): Finding to fix + **kwargs: Additional GCP-specific arguments (project_id, resource_id) + + Returns: + bool: True if fixing was successful, False otherwise + """ + try: + # Get values either from finding or kwargs + project_id = None + resource_id = None + + if finding: + project_id = ( + finding.project_id if hasattr(finding, "project_id") else None + ) + resource_id = ( + finding.resource_id if hasattr(finding, "resource_id") else None + ) + else: + project_id = kwargs.get("project_id") + resource_id = kwargs.get("resource_id") + + # Print the appropriate message based on available information + if project_id and resource_id: + print(f"\tFIXING {resource_id} in project {project_id}...") + elif project_id: + print(f"\tFIXING project {project_id}...") + elif resource_id: + print(f"\tFIXING Resource {resource_id}...") + else: + logger.error( + "Either finding or required kwargs (project_id, resource_id) must be provided" + ) + return False + + return True + + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return False diff --git a/prowler/providers/gcp/services/compute/compute_project_os_login_enabled/compute_project_os_login_enabled_fixer.py b/prowler/providers/gcp/services/compute/compute_project_os_login_enabled/compute_project_os_login_enabled_fixer.py new file mode 100644 index 0000000000..a30439b798 --- /dev/null +++ b/prowler/providers/gcp/services/compute/compute_project_os_login_enabled/compute_project_os_login_enabled_fixer.py @@ -0,0 +1,63 @@ +from typing import Optional + +from prowler.lib.check.models import Check_Report_GCP +from prowler.lib.logger import logger +from prowler.providers.gcp.lib.fix.fixer import GCPFixer +from prowler.providers.gcp.services.compute.compute_client import compute_client + + +class ComputeProjectOSLoginEnabledFixer(GCPFixer): + """ + Fixer for enabling OS Login at the project level. + This fixer enables the OS Login feature which provides centralized and automated SSH key pair management. + """ + + def __init__(self): + """ + Initialize Compute Engine fixer. + """ + super().__init__( + description="Enable OS Login at the project level", + cost_impact=False, + cost_description=None, + service="compute", + iam_policy_required={ + "roles": ["roles/compute.admin"], + }, + ) + + def fix(self, finding: Optional[Check_Report_GCP] = None, **kwargs) -> bool: + """ + Enable OS Login at the project level. + + Args: + finding (Optional[Check_Report_GCP]): Finding to fix + **kwargs: Additional arguments (project_id is required if finding is not provided) + + Returns: + bool: True if the operation is successful (OS Login is enabled), False otherwise + """ + try: + # Get project_id either from finding or kwargs + if finding: + project_id = finding.project_id + else: + project_id = kwargs.get("project_id") + + if not project_id: + raise ValueError("project_id is required") + + # Enable OS Login + request = compute_client.client.projects().setCommonInstanceMetadata( + project=project_id, + body={"items": [{"key": "enable-oslogin", "value": "TRUE"}]}, + ) + request.execute() + + return True + + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return False