diff --git a/prowler/providers/gcp/exceptions/exceptions.py b/prowler/providers/gcp/exceptions/exceptions.py index 65f6f27adc..5fe30484f6 100644 --- a/prowler/providers/gcp/exceptions/exceptions.py +++ b/prowler/providers/gcp/exceptions/exceptions.py @@ -37,6 +37,10 @@ class GCPBaseException(ProwlerException): "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.", }, + (1933, "GCPInvalidAccountCredentials"): { + "message": "Provider does not match with the expected project_id", + "remediation": "Check the provider and ensure it matches the expected project_id.", + }, } def __init__(self, code, file=None, original_exception=None, message=None): @@ -114,3 +118,10 @@ class GCPStaticCredentialsError(GCPCredentialsError): super().__init__( 1932, file=file, original_exception=original_exception, message=message ) + + +class GCPInvalidAccountCredentials(GCPBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 1933, file=file, original_exception=original_exception, message=message + ) diff --git a/prowler/providers/gcp/gcp_provider.py b/prowler/providers/gcp/gcp_provider.py index d9fb5d02d7..d56d281647 100644 --- a/prowler/providers/gcp/gcp_provider.py +++ b/prowler/providers/gcp/gcp_provider.py @@ -1,6 +1,7 @@ import os import re import sys +from typing import Optional from colorama import Fore, Style from google.auth import default, impersonated_credentials, load_credentials_from_dict @@ -18,6 +19,7 @@ from prowler.providers.gcp.exceptions.exceptions import ( GCPCloudResourceManagerAPINotUsedError, GCPGetProjectError, GCPHTTPError, + GCPInvalidAccountCredentials, GCPLoadCredentialsFromDictError, GCPNoAccesibleProjectsError, GCPSetUpSessionError, @@ -83,7 +85,7 @@ class GcpProvider(Provider): self._project_ids = [] self._projects = {} self._excluded_project_ids = [] - accessible_projects = self.get_projects() + accessible_projects = self.get_projects(self._session) if not accessible_projects: logger.critical("No Project IDs can be accessed via Google Credentials.") raise GCPNoAccesibleProjectsError( @@ -296,6 +298,7 @@ class GcpProvider(Provider): client_id: str = None, client_secret: str = None, refresh_token: str = None, + provider_id: Optional[str] = None, ) -> Connection: """ Test the connection to GCP with the provided credentials file or service account to impersonate. @@ -305,6 +308,11 @@ class GcpProvider(Provider): Args: credentials_file: str service_account: str + raise_on_exception: bool + client_id: str + client_secret: str + refresh_token: str + provider_id: Optional[str] -> The provider ID, for GCP it is the project ID Returns: Connection object with is_connected set to True if the connection is successful, or error set to the exception if the connection fails """ @@ -316,9 +324,15 @@ class GcpProvider(Provider): client_id, client_secret, refresh_token ) - session, _ = GcpProvider.setup_session( + session, project_id = GcpProvider.setup_session( credentials_file, service_account, gcp_credentials ) + if provider_id and project_id != provider_id: + # Logic to check if the provider ID matches the project ID + GcpProvider.validate_project_id( + provider_id=provider_id, credentials=session + ) + service = discovery.build("cloudresourcemanager", "v1", credentials=session) request = service.projects().list() request.execute() @@ -351,6 +365,12 @@ class GcpProvider(Provider): if raise_on_exception: raise http_error return Connection(error=http_error) + # Exceptions from validating Provider ID + except GCPInvalidAccountCredentials as not_valid_provider_id_error: + logger.critical(str(not_valid_provider_id_error)) + if raise_on_exception: + raise not_valid_provider_id_error + return Connection(error=not_valid_provider_id_error) except Exception as error: logger.critical( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" @@ -383,12 +403,13 @@ class GcpProvider(Provider): ) print_boxes(report_lines, report_title) - def get_projects(self) -> dict[str, GCPProject]: + @staticmethod + def get_projects(credentials) -> dict[str, GCPProject]: try: projects = {} service = discovery.build( - "cloudresourcemanager", "v1", credentials=self.session + "cloudresourcemanager", "v1", credentials=credentials ) request = service.projects().list() @@ -528,3 +549,34 @@ class GcpProvider(Provider): "refresh_token": refresh_token, "type": "authorized_user", } + + @staticmethod + def validate_project_id(provider_id: str, credentials: str = None) -> None: + """ + Validate the provider ID given the credentials, checking if the provider ID matches with the expected project_id using the method get_projects + + Args: + provider_id: str + credentials: str + + Returns: + None + + Raises: + GCPInvalidAccountCredentials if the provider ID does not match with the expected project_id + """ + + available_projects = list( + GcpProvider.get_projects(credentials=credentials).keys() + ) + + if len(available_projects) == 0: + raise GCPNoAccesibleProjectsError( + file=__file__, + message="No Project IDs can be accessed via Google Credentials.", + ) + elif provider_id not in available_projects: + raise GCPInvalidAccountCredentials( + file=__file__, + message="The provider ID does not match with the expected project_id.", + ) diff --git a/tests/providers/gcp/gcp_provider_test.py b/tests/providers/gcp/gcp_provider_test.py index 45c2fa1463..e84ab6fba5 100644 --- a/tests/providers/gcp/gcp_provider_test.py +++ b/tests/providers/gcp/gcp_provider_test.py @@ -11,7 +11,11 @@ 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.common.models import Connection +from prowler.providers.gcp.exceptions.exceptions import ( + GCPInvalidAccountCredentials, + GCPTestConnectionError, +) from prowler.providers.gcp.gcp_provider import GcpProvider from prowler.providers.gcp.models import GCPIdentityInfo, GCPProject @@ -538,3 +542,67 @@ class TestGCPProvider: ) assert e.type == GCPTestConnectionError assert "Test exception" in e.value.args[0] + + def test_test_connection_valid_project_id(self): + project_id = "test-project-id" + mocked_service = MagicMock() + + mocked_service.projects.get.return_value = MagicMock( + execute=MagicMock(return_value={"projectId": project_id}) + ) + + with patch( + "prowler.providers.gcp.gcp_provider.GcpProvider.setup_session", + return_value=(None, project_id), + ), patch( + "prowler.providers.gcp.gcp_provider.discovery.build", + return_value=mocked_service, + ): + output = GcpProvider.test_connection( + client_id="test-client-id", + client_secret="test-client-secret", + refresh_token="test-refresh-token", + provider_id=project_id, + ) + assert Connection(is_connected=True, error=None) == output + + def test_test_connection_invalid_project_id(self): + mocked_service = MagicMock() + + projects = { + "test-valid-project": GCPProject( + number="55555555", + id="project/55555555", + name="test-project", + labels={"test": "value"}, + lifecycle_state="", + ), + } + + mocked_service.projects.get.return_value = MagicMock( + execute=MagicMock(return_value={"projects": projects}) + ) + + with patch( + "prowler.providers.gcp.gcp_provider.GcpProvider.setup_session", + return_value=(None, "test-valid-project"), + ), patch( + "prowler.providers.gcp.gcp_provider.discovery.build", + return_value=mocked_service, + ), patch( + "prowler.providers.gcp.gcp_provider.GcpProvider.validate_project_id" + ) as mock_validate_project_id: + + mock_validate_project_id.side_effect = GCPInvalidAccountCredentials( + "Invalid project ID" + ) + + with pytest.raises(Exception) as e: + GcpProvider.test_connection( + client_id="test-client-id", + client_secret="test-client-secret", + refresh_token="test-refresh-token", + provider_id="test-invalid-project", + ) + + assert e.type == GCPInvalidAccountCredentials