ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 20:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(gcp): ignore redirect balancers and add regional ones (#7449)

Browse Source 
Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>
This commit is contained in:
Prowler Bot
2025-04-07 18:43:45 +02:00
committed by GitHub
parent 611681488d
commit f52929673d
5 changed files with 123 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/gcp/services/compute/compute_loadbalancer_logging_enabled/compute_loadbalancer_logging_enabled.py
+14 -12
View File
@@ -6,18 +6,20 @@ class compute_loadbalancer_logging_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for lb in compute_client.load_balancers:
report = Check_Report_GCP(
metadata=self.metadata(),
resource=lb,
location=compute_client.region,
)
report.status = "PASS"
report.status_extended = f"LoadBalancer {lb.name} has logging enabled."
if not lb.logging:
report.status = "FAIL"
report.status_extended = (
f"LoadBalancer {lb.name} does not have logging enabled."
# Only load balancers with backend service can have logging enabled
if lb.service:
report = Check_Report_GCP(
metadata=self.metadata(),
resource=lb,
location=compute_client.region,
)
findings.append(report)
report.status = "PASS"
report.status_extended = f"LoadBalancer {lb.name} has logging enabled."
if not lb.logging:
report.status = "FAIL"
report.status_extended = (
f"LoadBalancer {lb.name} does not have logging enabled."
)
findings.append(report)
return findings
prowler/providers/gcp/services/compute/compute_service.py
+50 -9
View File
@@ -17,10 +17,10 @@ class Compute(GCPService):
self.firewalls = []
self.compute_projects = []
self.load_balancers = []
self._get_url_maps()
self._describe_backend_service()
self._get_regions()
self._get_projects()
self._get_url_maps()
self._describe_backend_service()
self._get_zones()
self.__threading_call__(self._get_instances, self.zones)
self._get_networks()
@@ -260,6 +260,7 @@ class Compute(GCPService):
def _get_url_maps(self):
for project_id in self.project_ids:
try:
# Global URL maps
request = self.client.urlMaps().list(project=project_id)
while request is not None:
response = request.execute()
@@ -280,19 +281,59 @@ class Compute(GCPService):
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
try:
# Regional URL maps
for region in self.regions:
request = self.client.regionUrlMaps().list(
project=project_id, region=region
)
while request is not None:
response = request.execute()
for urlmap in response.get("items", []):
self.load_balancers.append(
LoadBalancer(
name=urlmap["name"],
id=urlmap["id"],
service=urlmap.get("defaultService", ""),
project_id=project_id,
)
)
request = self.client.regionUrlMaps().list_next(
previous_request=request, previous_response=response
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _describe_backend_service(self):
for balancer in self.load_balancers:
if balancer.service:
try:
response = (
self.client.backendServices()
.get(
project=balancer.project_id,
backendService=balancer.service.split("/")[-1],
backend_service_name = balancer.service.split("/")[-1]
is_regional = "/regions/" in balancer.service
if is_regional:
region = balancer.service.split("/regions/")[1].split("/")[0]
response = (
self.client.regionBackendServices()
.get(
project=balancer.project_id,
region=region,
backendService=backend_service_name,
)
.execute()
)
.execute()
)
else:
response = (
self.client.backendServices()
.get(
project=balancer.project_id,
backendService=backend_service_name,
)
.execute()
)
balancer.logging = response.get("logConfig", {}).get(
"enable", False
)
tests/providers/gcp/gcp_fixtures.py
+16
View File
@@ -887,6 +887,22 @@ def mock_api_urlMaps_calls(client: MagicMock):
}
client.urlMaps().list_next.return_value = None
client.regionUrlMaps().list().execute.return_value = {
"items": [
{
"name": "regional_url_map1",
"id": str(uuid4()),
"defaultService": "regional_service1",
},
{
"name": "regional_url_map2",
"id": str(uuid4()),
"defaultService": "regional_service2",
},
]
}
client.regionUrlMaps().list_next.return_value = None
client.backendServices().get().execute.side_effect = [
{
"logConfig": {"enable": True},
tests/providers/gcp/services/compute/compute_loadbalancer_logging_enabled/compute_loadbalancer_logging_enabled_test.py
+31
View File
@@ -106,3 +106,34 @@ class Test_compute_loadbalancer_logging_enabled:
assert result[0].resource_name == load_balancer.name
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == compute_client.region
def test_one_load_balancer_without_backend_service(self):
from prowler.providers.gcp.services.compute.compute_service import LoadBalancer
load_balancer = LoadBalancer(
name="test", id="test_id", project_id=GCP_PROJECT_ID, service=""
)
compute_client = mock.MagicMock()
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.load_balancers = [load_balancer]
compute_client.region = "global"
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.compute.compute_loadbalancer_logging_enabled.compute_loadbalancer_logging_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_loadbalancer_logging_enabled.compute_loadbalancer_logging_enabled import (
compute_loadbalancer_logging_enabled,
)
check = compute_loadbalancer_logging_enabled()
result = check.execute()
assert len(result) == 0
tests/providers/gcp/services/compute/compute_service_test.py
+12 -7
View File
@@ -156,19 +156,24 @@ class TestComputeService:
assert compute_client.firewalls[2].direction == "INGRESS"
assert compute_client.firewalls[2].project_id == GCP_PROJECT_ID
assert len(compute_client.load_balancers) == 2
assert len(compute_client.load_balancers) == 4
assert compute_client.load_balancers[0].name == "url_map1"
assert compute_client.load_balancers[0].id.__class__.__name__ == "str"
assert compute_client.load_balancers[0].service == "service1"
assert compute_client.load_balancers[0].project_id == GCP_PROJECT_ID
assert compute_client.load_balancers[0].logging
assert compute_client.load_balancers[1].name == "url_map2"
assert compute_client.load_balancers[1].id.__class__.__name__ == "str"
assert compute_client.load_balancers[1].service == "service2"
assert compute_client.load_balancers[1].project_id == GCP_PROJECT_ID
assert len(compute_client.load_balancers) == 2
assert compute_client.load_balancers[0].logging
assert not compute_client.load_balancers[1].logging
assert compute_client.load_balancers[2].name == "regional_url_map1"
assert compute_client.load_balancers[2].id.__class__.__name__ == "str"
assert compute_client.load_balancers[2].service == "regional_service1"
assert compute_client.load_balancers[2].project_id == GCP_PROJECT_ID
assert not compute_client.load_balancers[2].logging
assert compute_client.load_balancers[3].name == "regional_url_map2"
assert compute_client.load_balancers[3].id.__class__.__name__ == "str"
assert compute_client.load_balancers[3].service == "regional_service2"
assert compute_client.load_balancers[3].project_id == GCP_PROJECT_ID
assert not compute_client.load_balancers[3].logging