refactor(WAF): Rename WAF to WAFRegional and Add Global WAF Service (#5389)

Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Hugo Pereira Brito
2024-10-14 17:49:26 +02:00
committed by GitHub
parent 037e40f8e4
commit 2c0c1f7d09
5 changed files with 103 additions and 17 deletions
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
from prowler.providers.aws.services.waf.waf_client import waf_client
from prowler.providers.aws.services.waf.wafregional_client import wafregional_client
from prowler.providers.aws.services.wafv2.wafv2_client import wafv2_client
@@ -22,7 +22,7 @@ class elbv2_waf_acl_attached(Check):
if lb_arn in acl.albs:
report.status = "PASS"
report.status_extended = f"ELBv2 ALB {lb.name} is protected by WAFv2 Web ACL {acl.name}."
for acl in waf_client.web_acls.values():
for acl in wafregional_client.web_acls.values():
if lb_arn in acl.albs:
report.status = "PASS"
report.status_extended = f"ELBv2 ALB {lb.name} is protected by WAFv1 Web ACL {acl.name}."
@@ -6,6 +6,52 @@ from prowler.providers.aws.lib.service.service import AWSService
class WAF(AWSService):
def __init__(self, provider):
# Call AWSService's __init__
super().__init__("waf", provider)
self.web_acls = {}
self.__threading_call__(self._list_web_acls)
self.__threading_call__(
self._list_resources_for_web_acl, self.web_acls.values()
)
def _list_web_acls(self, regional_client):
logger.info("WAF - Listing Regional Web ACLs...")
try:
for waf in regional_client.list_web_acls()["WebACLs"]:
if not self.audit_resources or (
is_resource_filtered(waf["WebACLId"], self.audit_resources)
):
arn = f"arn:aws:waf:{regional_client.region}:{self.audited_account}:webacl/{waf['WebACLId']}"
self.web_acls[arn] = WebAcl(
arn=arn,
name=waf["Name"],
id=waf["WebACLId"],
albs=[],
region=regional_client.region,
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _list_resources_for_web_acl(self, regional_client):
logger.info("WAF - Describing resources...")
try:
for acl in self.web_acls.values():
if acl.region == regional_client.region:
for resource in regional_client.list_resources_for_web_acl(
WebACLId=acl.id, ResourceType="APPLICATION_LOAD_BALANCER"
)["ResourceArns"]:
acl.albs.append(resource)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class WAFRegional(AWSService):
def __init__(self, provider):
# Call AWSService's __init__
super().__init__("waf-regional", provider)
@@ -50,6 +96,8 @@ class WAF(AWSService):
class WebAcl(BaseModel):
"""Web ACL Model for WAF and WAFRegional"""
arn: str
name: str
id: str
@@ -0,0 +1,4 @@
from prowler.providers.aws.services.waf.waf_service import WAFRegional
from prowler.providers.common.provider import Provider
wafregional_client = WAFRegional(Provider.get_global_provider())
@@ -39,7 +39,7 @@ class Test_elbv2_waf_acl_attached:
@mock_aws
def test_elb_no_balancers(self):
from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
from prowler.providers.aws.services.waf.waf_service import WAF
from prowler.providers.aws.services.waf.waf_service import WAFRegional
from prowler.providers.aws.services.wafv2.wafv2_service import WAFv2
with mock.patch(
@@ -58,8 +58,8 @@ class Test_elbv2_waf_acl_attached:
set_mocked_aws_provider([AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1])
),
), mock.patch(
"prowler.providers.aws.services.elbv2.elbv2_waf_acl_attached.elbv2_waf_acl_attached.waf_client",
new=WAF(
"prowler.providers.aws.services.elbv2.elbv2_waf_acl_attached.elbv2_waf_acl_attached.wafregional_client",
new=WAFRegional(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1])
),
):
@@ -112,7 +112,7 @@ class Test_elbv2_waf_acl_attached:
)["LoadBalancers"][0]
from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
from prowler.providers.aws.services.waf.waf_service import WAF
from prowler.providers.aws.services.waf.waf_service import WAFRegional
from prowler.providers.aws.services.wafv2.wafv2_service import WAFv2
with mock.patch(
@@ -131,8 +131,8 @@ class Test_elbv2_waf_acl_attached:
set_mocked_aws_provider([AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1])
),
), mock.patch(
"prowler.providers.aws.services.elbv2.elbv2_waf_acl_attached.elbv2_waf_acl_attached.waf_client",
new=WAF(
"prowler.providers.aws.services.elbv2.elbv2_waf_acl_attached.elbv2_waf_acl_attached.wafregional_client",
new=WAFRegional(
set_mocked_aws_provider([AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1])
),
):
@@ -194,7 +194,7 @@ class Test_elbv2_waf_acl_attached:
wafv2.associate_web_acl(WebACLArn=waf["ARN"], ResourceArn=lb["LoadBalancerArn"])
from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
from prowler.providers.aws.services.waf.waf_service import WAF
from prowler.providers.aws.services.waf.waf_service import WAFRegional
from prowler.providers.aws.services.wafv2.wafv2_service import WAFv2
with mock.patch(
@@ -214,8 +214,8 @@ class Test_elbv2_waf_acl_attached:
),
) as service_client:
with mock.patch(
"prowler.providers.aws.services.elbv2.elbv2_waf_acl_attached.elbv2_waf_acl_attached.waf_client",
new=WAF(
"prowler.providers.aws.services.elbv2.elbv2_waf_acl_attached.elbv2_waf_acl_attached.wafregional_client",
new=WAFRegional(
set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
@@ -2,7 +2,7 @@ from unittest.mock import patch
import botocore
from prowler.providers.aws.services.waf.waf_service import WAF
from prowler.providers.aws.services.waf.waf_service import WAF, WAFRegional
from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider
# Mocking WAF-Regional Calls
@@ -43,12 +43,13 @@ def mock_generate_regional_clients(provider, service):
new=mock_generate_regional_clients,
)
class Test_WAF_Service:
# Test WAF Service
def test_service(self):
# WAF client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
waf = WAF(aws_provider)
assert waf.service == "waf-regional"
assert waf.service == "waf"
# Test WAF Client
def test_client(self):
@@ -56,7 +57,7 @@ class Test_WAF_Service:
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
waf = WAF(aws_provider)
for regional_client in waf.regional_clients.values():
assert regional_client.__class__.__name__ == "WAFRegional"
assert regional_client.__class__.__name__ == "WAF"
# Test WAF Session
def test__get_session__(self):
@@ -70,18 +71,51 @@ class Test_WAF_Service:
# WAF client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
waf = WAF(aws_provider)
waf_arn = "arn:aws:waf-regional:eu-west-1:123456789012:webacl/my-web-acl-id"
waf_arn = "arn:aws:waf:eu-west-1:123456789012:webacl/my-web-acl-id"
assert len(waf.web_acls) == 1
assert waf.web_acls[waf_arn].name == "my-web-acl"
assert waf.web_acls[waf_arn].region == AWS_REGION_EU_WEST_1
assert waf.web_acls[waf_arn].id == "my-web-acl-id"
# Test WAF Describe Web ACLs Resources
# Test WAFRegional Describe Web ACLs Resources
def test_list_resources_for_web_acl(self):
# WAF client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
waf = WAF(aws_provider)
waf = WAFRegional(aws_provider)
waf_arn = "arn:aws:waf-regional:eu-west-1:123456789012:webacl/my-web-acl-id"
assert len(waf.web_acls) == 1
assert len(waf.web_acls[waf_arn].albs) == 1
assert "alb-arn" in waf.web_acls[waf_arn].albs
# Test WAFRegional Service
def test_service_regional(self):
# WAF client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
waf = WAFRegional(aws_provider)
assert waf.service == "waf-regional"
# Test WAFRegional Client
def test_client_regional(self):
# WAF client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
waf = WAFRegional(aws_provider)
for regional_client in waf.regional_clients.values():
assert regional_client.__class__.__name__ == "WAFRegional"
# Test WAFRegional Session
def test__get_session___regional(self):
# WAF client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
waf = WAFRegional(aws_provider)
assert waf.session.__class__.__name__ == "Session"
# Test WAFRegional Describe Web ACLs
def test_list_web_acls_waf_regional(self):
# WAF client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
waf = WAFRegional(aws_provider)
waf_arn = "arn:aws:waf-regional:eu-west-1:123456789012:webacl/my-web-acl-id"
assert len(waf.web_acls) == 1
assert waf.web_acls[waf_arn].name == "my-web-acl"
assert waf.web_acls[waf_arn].region == AWS_REGION_EU_WEST_1
assert waf.web_acls[waf_arn].id == "my-web-acl-id"