feat(gcp): add static credentials for gcp provider (#5364)

prowler/providers/gcp/exceptions/exceptions.py

@@ -29,6 +29,14 @@ class GCPBaseException(ProwlerException):
             "message": "Error testing connection to GCP",
             "remediation": "Check the connection and ensure it is properly set up.",
         },
+        (1931, "GCPLoadCredentialsFromDictError"): {
+            "message": "Error loading credentials from dictionary",
+            "remediation": "Check the credentials and ensure they are properly set up. client_id, client_secret and refresh_token are required.",
+        },
+        (1932, "GCPStaticCredentialsError"): {
+            "message": "Error loading static credentials",
+            "remediation": "Check the credentials and ensure they are properly set up. client_id, client_secret and refresh_token are required.",
+        },
     }
 
     def __init__(self, code, file=None, original_exception=None, message=None):
@@ -45,6 +53,13 @@ class GCPBaseException(ProwlerException):
         )
 
 
+class GCPCredentialsError(GCPBaseException):
+    """Base class for GCP credentials errors."""
+
+    def __init__(self, code, file=None, original_exception=None, message=None):
+        super().__init__(code, file, original_exception, message)
+
+
 class GCPCloudResourceManagerAPINotUsedError(GCPBaseException):
     def __init__(self, file=None, original_exception=None, message=None):
         super().__init__(
@@ -59,21 +74,21 @@ class GCPHTTPError(GCPBaseException):
         )
 
 
-class GCPNoAccesibleProjectsError(GCPBaseException):
+class GCPNoAccesibleProjectsError(GCPCredentialsError):
     def __init__(self, file=None, original_exception=None, message=None):
         super().__init__(
             1927, file=file, original_exception=original_exception, message=message
         )
 
 
-class GCPSetUpSessionError(GCPBaseException):
+class GCPSetUpSessionError(GCPCredentialsError):
     def __init__(self, file=None, original_exception=None, message=None):
         super().__init__(
             1928, file=file, original_exception=original_exception, message=message
         )
 
 
-class GCPGetProjectError(GCPBaseException):
+class GCPGetProjectError(GCPCredentialsError):
     def __init__(self, file=None, original_exception=None, message=None):
         super().__init__(
             1929, file=file, original_exception=original_exception, message=message
@@ -85,3 +100,17 @@ class GCPTestConnectionError(GCPBaseException):
         super().__init__(
             1930, file=file, original_exception=original_exception, message=message
         )
+
+
+class GCPLoadCredentialsFromDictError(GCPCredentialsError):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1931, file=file, original_exception=original_exception, message=message
+        )
+
+
+class GCPStaticCredentialsError(GCPCredentialsError):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1932, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/gcp/gcp_provider.py

@@ -3,7 +3,7 @@ import re
 import sys
 
 from colorama import Fore, Style
-from google.auth import default, impersonated_credentials
+from google.auth import default, impersonated_credentials, load_credentials_from_dict
 from google.auth.transport.requests import Request
 from google.oauth2.credentials import Credentials
 from googleapiclient import discovery
@@ -18,8 +18,10 @@ from prowler.providers.gcp.exceptions.exceptions import (
     GCPCloudResourceManagerAPINotUsedError,
     GCPGetProjectError,
     GCPHTTPError,
+    GCPLoadCredentialsFromDictError,
     GCPNoAccesibleProjectsError,
     GCPSetUpSessionError,
+    GCPStaticCredentialsError,
     GCPTestConnectionError,
 )
 from prowler.providers.gcp.lib.mutelist.mutelist import GCPMutelist
@@ -46,6 +48,9 @@ class GcpProvider(Provider):
         list_project_ids: bool = False,
         audit_config: dict = {},
         fixer_config: dict = {},
+        client_id: str = None,
+        client_secret: str = None,
+        refresh_token: str = None,
     ):
         """
         GCP Provider constructor
@@ -58,12 +63,21 @@ class GcpProvider(Provider):
             list_project_ids: bool
             audit_config: dict
             fixer_config: dict
+            client_id: str
+            client_secret: str
+            refresh_token: str
         """
         logger.info("Instantiating GCP Provider ...")
         self._impersonated_service_account = impersonate_service_account
+        # Set the GCP credentials using the provided client_id, client_secret and refresh_token
+        gcp_credentials = None
+        if any([client_id, client_secret, refresh_token]):
+            gcp_credentials = self.validate_static_arguments(
+                client_id, client_secret, refresh_token
+            )
 
         self._session, self._default_project_id = self.setup_session(
-            credentials_file, self._impersonated_service_account
+            credentials_file, self._impersonated_service_account, gcp_credentials
         )
 
         self._project_ids = []
@@ -212,7 +226,9 @@ class GcpProvider(Provider):
         }
 
     @staticmethod
-    def setup_session(credentials_file: str, service_account: str) -> tuple:
+    def setup_session(
+        credentials_file: str, service_account: str, gcp_credentials: dict = None
+    ) -> tuple:
         """
         Setup the GCP session with the provided credentials file or service account to impersonate
         Args:
@@ -224,6 +240,22 @@ class GcpProvider(Provider):
         try:
             scopes = ["https://www.googleapis.com/auth/cloud-platform"]
 
+            if gcp_credentials:
+                logger.info("Using GCP static credentials")
+                logger.info("GCP provider: Setting credentials from dict...")
+                try:
+                    credentials, default_project_id = load_credentials_from_dict(
+                        info=gcp_credentials
+                    )
+                    return credentials, default_project_id
+                except Exception as error:
+                    logger.critical(
+                        f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                    )
+                    raise GCPLoadCredentialsFromDictError(
+                        file=__file__, original_exception=error
+                    )
+
             if credentials_file:
                 logger.info(f"Using credentials file: {credentials_file}")
                 logger.info(
@@ -261,6 +293,9 @@ class GcpProvider(Provider):
         credentials_file: str = None,
         service_account: str = None,
         raise_on_exception: bool = True,
+        client_id: str = None,
+        client_secret: str = None,
+        refresh_token: str = None,
     ) -> Connection:
         """
         Test the connection to GCP with the provided credentials file or service account to impersonate.
@@ -274,13 +309,27 @@ class GcpProvider(Provider):
             Connection object with is_connected set to True if the connection is successful, or error set to the exception if the connection fails
         """
         try:
-            session, _ = GcpProvider.setup_session(credentials_file, service_account)
+            # Set the GCP credentials using the provided client_id, client_secret and refresh_token
+            gcp_credentials = None
+            if any([client_id, client_secret, refresh_token]):
+                gcp_credentials = GcpProvider.validate_static_arguments(
+                    client_id, client_secret, refresh_token
+                )
+
+            session, _ = GcpProvider.setup_session(
+                credentials_file, service_account, gcp_credentials
+            )
             service = discovery.build("cloudresourcemanager", "v1", credentials=session)
             request = service.projects().list()
             request.execute()
             return Connection(is_connected=True)
 
         # Errors from setup_session
+        except GCPLoadCredentialsFromDictError as load_credentials_error:
+            logger.critical(str(load_credentials_error))
+            if raise_on_exception:
+                raise load_credentials_error
+            return Connection(error=load_credentials_error)
         except GCPSetUpSessionError as setup_session_error:
             logger.critical(str(setup_session_error))
             if raise_on_exception:
@@ -447,3 +496,35 @@ class GcpProvider(Provider):
                 project_to_match,
             )
         ) or input_project == project_to_match
+
+    @staticmethod
+    def validate_static_arguments(
+        client_id: str = None, client_secret: str = None, refresh_token: str = None
+    ) -> dict:
+        """
+        Validate the static arguments client_id, client_secret and refresh_token
+
+        Args:
+            client_id: str
+            client_secret: str
+            refresh_token: str
+
+        Returns:
+            dict
+
+        Raises:
+            GCPStaticCredentialsError if any of the static arguments is missing
+        """
+
+        if not client_id or not client_secret or not refresh_token:
+            raise GCPStaticCredentialsError(
+                file=__file__,
+                message="client_id, client_secret and refresh_token are required.",
+            )
+
+        return {
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "refresh_token": refresh_token,
+            "type": "authorized_user",
+        }

tests/providers/gcp/gcp_provider_test.py

@@ -2,6 +2,7 @@ from argparse import Namespace
 from datetime import datetime
 from os import environ
 
+import pytest
 from freezegun import freeze_time
 from mock import MagicMock, patch
 
@@ -10,6 +11,7 @@ from prowler.config.config import (
     default_fixer_config_file_path,
     load_and_validate_config_file,
 )
+from prowler.providers.gcp.exceptions.exceptions import GCPTestConnectionError
 from prowler.providers.gcp.gcp_provider import GcpProvider
 from prowler.providers.gcp.models import GCPIdentityInfo, GCPProject
 
@@ -25,6 +27,9 @@ class TestGCPProvider:
         fixer_config = load_and_validate_config_file(
             "gcp", default_fixer_config_file_path
         )
+        client_id = "test-client-id"
+        client_secret = "test-client-secret"
+        refresh_token = "test-refresh-token"
 
         projects = {
             "test-project": GCPProject(
@@ -63,6 +68,9 @@ class TestGCPProvider:
                 list_project_id,
                 audit_config=audit_config,
                 fixer_config=fixer_config,
+                client_id=client_id,
+                client_secret=client_secret,
+                refresh_token=refresh_token,
             )
             assert gcp_provider.session is None
             assert gcp_provider.project_ids == ["test-project"]
@@ -127,6 +135,9 @@ class TestGCPProvider:
                 arguments.list_project_id,
                 arguments.config_file,
                 arguments.fixer_config,
+                client_id="test-client-id",
+                client_secret="test-client-secret",
+                refresh_token="test-refresh-token",
             )
 
             input_project = "sys-*"
@@ -199,6 +210,9 @@ class TestGCPProvider:
                 arguments.list_project_id,
                 arguments.config_file,
                 arguments.fixer_config,
+                client_id=None,
+                client_secret=None,
+                refresh_token=None,
             )
             assert environ["GOOGLE_APPLICATION_CREDENTIALS"] == "test_credentials_file"
             assert gcp_provider.session is not None
@@ -258,6 +272,9 @@ class TestGCPProvider:
                 arguments.list_project_id,
                 arguments.config_file,
                 arguments.fixer_config,
+                client_id=None,
+                client_secret=None,
+                refresh_token=None,
             )
             assert environ["GOOGLE_APPLICATION_CREDENTIALS"] == "test_credentials_file"
             assert gcp_provider.session is not None
@@ -325,6 +342,9 @@ class TestGCPProvider:
                 arguments.list_project_id,
                 arguments.config_file,
                 arguments.fixer_config,
+                client_id=None,
+                client_secret=None,
+                refresh_token=None,
             )
             gcp_provider.print_credentials()
             captured = capsys.readouterr()
@@ -391,6 +411,9 @@ class TestGCPProvider:
                 arguments.list_project_id,
                 arguments.config_file,
                 arguments.fixer_config,
+                client_id=None,
+                client_secret=None,
+                refresh_token=None,
             )
             gcp_provider.print_credentials()
             captured = capsys.readouterr()
@@ -465,6 +488,9 @@ class TestGCPProvider:
                 arguments.list_project_id,
                 arguments.config_file,
                 arguments.fixer_config,
+                client_id=None,
+                client_secret=None,
+                refresh_token=None,
             )
             gcp_provider.print_credentials()
             captured = capsys.readouterr()
@@ -479,3 +505,36 @@ class TestGCPProvider:
                 "Excluded GCP Project IDs:" in captured.out
                 and "test-excluded-project" in captured.out
             )
+
+    def test_init_only_client_id(self):
+        with pytest.raises(Exception) as e:
+            GcpProvider(client_id="test-client-id")
+        assert "client_secret and refresh_token are required" in e.value.args[0]
+
+    def test_validate_static_arguments(self):
+        output = GcpProvider.validate_static_arguments(
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            refresh_token="test-refresh-token",
+        )
+
+        assert output == {
+            "client_id": "test-client-id",
+            "client_secret": "test-client-secret",
+            "refresh_token": "test-refresh-token",
+            "type": "authorized_user",
+        }
+
+    def test_test_connection_with_exception(self):
+        with patch(
+            "prowler.providers.gcp.gcp_provider.GcpProvider.setup_session",
+            side_effect=Exception("Test exception"),
+        ):
+            with pytest.raises(Exception) as e:
+                GcpProvider.test_connection(
+                    client_id="test-client-id",
+                    client_secret="test-client-secret",
+                    refresh_token="test-refresh-token",
+                )
+            assert e.type == GCPTestConnectionError
+            assert "Test exception" in e.value.args[0]