feat(wafv2): Add check wafv2_webacl_logging_enabled (#2898)

Co-authored-by: Pepe Fagoaga <pepe@verica.io>
Co-authored-by: Sergio Garcia <sergargar1@gmail.com>
This commit is contained in:
DevOpSpace
2023-10-04 13:10:47 +04:00
committed by GitHub
parent a4d3e78eb1
commit 4c15318f28
5 changed files with 190 additions and 7 deletions

View File

@@ -13,6 +13,7 @@ class WAFv2(AWSService):
self.web_acls = []
self.__threading_call__(self.__list_web_acls__)
self.__threading_call__(self.__list_resources_for_web_acl__)
self.__threading_call__(self.__get_logging_configuration__)
def __list_web_acls__(self, regional_client):
logger.info("WAFv2 - Listing Regional Web ACLs...")
@@ -35,20 +36,36 @@ class WAFv2(AWSService):
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def __get_logging_configuration__(self, regional_client):
logger.info("WAFv2 - Get Logging Configuration...")
for acl in self.web_acls:
if acl.region == regional_client.region:
try:
logging_enabled = regional_client.get_logging_configuration(
ResourceArn=acl.arn
)
acl.logging_enabled = bool(
logging_enabled["LoggingConfiguration"]["LogDestinationConfigs"]
)
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("WAFv2 - Describing resources...")
try:
for acl in self.web_acls:
if acl.region == regional_client.region:
for acl in self.web_acls:
if acl.region == regional_client.region:
try:
for resource in regional_client.list_resources_for_web_acl(
WebACLArn=acl.arn, 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}"
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class WebAclv2(BaseModel):
@@ -57,3 +74,4 @@ class WebAclv2(BaseModel):
id: str
albs: list[str]
region: str
logging_enabled: bool = False

View File

@@ -0,0 +1,34 @@
{
"Provider": "aws",
"CheckID": "wafv2_webacl_logging_enabled",
"CheckTitle": "Check if AWS WAFv2 WebACL logging is enabled",
"CheckType": [
"Logging and Monitoring"
],
"ServiceName": "wafv2",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:wafv2:region:account-id:webacl/webacl-id",
"Severity": "medium",
"ResourceType": "AwsWafv2WebAcl",
"Description": "Check if AWS WAFv2 logging is enabled",
"Risk": "Enabling AWS WAFv2 logging helps monitor and analyze traffic patterns for enhanced security.",
"RelatedUrl": "https://docs.aws.amazon.com/waf/latest/developerguide/logging.html",
"Remediation": {
"Code": {
"CLI": "aws wafv2 update-web-acl-logging-configuration --scope REGIONAL --web-acl-arn arn:partition:wafv2:region:account-id:webacl/webacl-id --logging-configuration '{\"LogDestinationConfigs\": [\"arn:partition:logs:region:account-id:log-group:log-group-name\"]}'",
"NativeIaC": "",
"Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/WAF/enable-web-acls-logging.html",
"Terraform": "https://docs.bridgecrew.io/docs/bc_aws_logging_33#terraform"
},
"Recommendation": {
"Text": "Enable AWS WAFv2 logging for your Web ACLs to monitor and analyze traffic patterns effectively.",
"Url": "https://docs.aws.amazon.com/waf/latest/developerguide/logging.html"
}
},
"Categories": [
"logging"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}

View File

@@ -0,0 +1,27 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.wafv2.wafv2_client import wafv2_client
class wafv2_webacl_logging_enabled(Check):
def execute(self):
findings = []
for web_acl in wafv2_client.web_acls:
report = Check_Report_AWS(self.metadata())
report.region = web_acl.region
report.resource_id = web_acl.id
report.resource_arn = web_acl.arn
if web_acl.logging_enabled:
report.status = "PASS"
report.status_extended = (
f"AWS WAFv2 Web ACL {web_acl.id} has logging enabled."
)
else:
report.status = "FAIL"
report.status_extended = (
f"AWS WAFv2 Web ACL {web_acl.id} does not have logging enabled."
)
findings.append(report)
return findings

View File

@@ -0,0 +1,104 @@
from unittest import mock
from uuid import uuid4
from prowler.providers.aws.services.wafv2.wafv2_service import WebAclv2
AWS_REGION = "eu-west-1"
AWS_ACCOUNT_NUMBER = "123456789012"
waf_id = str(uuid4())
waf_name = "waf-example"
waf_arn = f"arn:aws:wafv2:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:regional/webacl/{waf_name}/{waf_id}"
class Test_wafv2_webacl_logging_enabled:
def test_no_web_acls(self):
wafv2_client = mock.MagicMock
wafv2_client.web_acls = []
with mock.patch(
"prowler.providers.aws.services.wafv2.wafv2_service.WAFv2",
new=wafv2_client,
), mock.patch(
"prowler.providers.aws.services.wafv2.wafv2_client.wafv2_client",
new=wafv2_client,
):
from prowler.providers.aws.services.wafv2.wafv2_webacl_logging_enabled.wafv2_webacl_logging_enabled import (
wafv2_webacl_logging_enabled,
)
check = wafv2_webacl_logging_enabled()
result = check.execute()
assert len(result) == 0
def test_wafv2_wb_acl_with_logging(self):
wafv2_client = mock.MagicMock
wafv2_client.web_acls = []
wafv2_client.enabled = True
wafv2_client.web_acls.append(
WebAclv2(
arn=waf_arn,
name=waf_name,
id=waf_id,
albs=[],
region=AWS_REGION,
logging_enabled=True,
)
)
with mock.patch(
"prowler.providers.aws.services.wafv2.wafv2_service.WAFv2",
new=wafv2_client,
), mock.patch(
"prowler.providers.aws.services.wafv2.wafv2_client.wafv2_client",
new=wafv2_client,
):
from prowler.providers.aws.services.wafv2.wafv2_webacl_logging_enabled.wafv2_webacl_logging_enabled import (
wafv2_webacl_logging_enabled,
)
check = wafv2_webacl_logging_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"AWS WAFv2 Web ACL {waf_id} has logging enabled."
)
assert result[0].resource_id == waf_id
assert result[0].resource_arn == waf_arn
assert result[0].region == AWS_REGION
def test_wafv2_wb_acl_without_logging(self):
wafv2_client = mock.MagicMock
wafv2_client.web_acls = []
wafv2_client.enabled = True
wafv2_client.web_acls.append(
WebAclv2(
arn=waf_arn,
name=waf_name,
id=waf_id,
albs=[],
region=AWS_REGION,
logging_enabled=False,
)
)
with mock.patch(
"prowler.providers.aws.services.wafv2.wafv2_service.WAFv2",
new=wafv2_client,
), mock.patch(
"prowler.providers.aws.services.wafv2.wafv2_client.wafv2_client",
new=wafv2_client,
):
from prowler.providers.aws.services.wafv2.wafv2_webacl_logging_enabled.wafv2_webacl_logging_enabled import (
wafv2_webacl_logging_enabled,
)
check = wafv2_webacl_logging_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"AWS WAFv2 Web ACL {waf_id} does not have logging enabled."
)
assert result[0].resource_id == waf_id
assert result[0].resource_arn == waf_arn
assert result[0].region == AWS_REGION