pmh_only/bp-check
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add feature: application load balancer bp check

Browse Source
This commit is contained in:
Donggeun Lee 2024-08-05 15:09:59 +09:00 committed by skyuecx0630
parent f9cc376f39
commit a06954f523
1 changed files with 95 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
services/alb.py
View File

@ -2,36 +2,122 @@ from models import RuleCheckResult
import boto3
# client = boto3.client("")
client = boto3.client("elbv2")
wafv2_client = boto3.client("wafv2")
def alb_http_drop_invalid_header_enabled():
load_balancers = client.describe_load_balancers()
compliant_resource = []
non_compliant_resources = []
for load_balancer in load_balancers['LoadBalancers']:
response = client.describe_load_balancer_attributes(
LoadBalancerArn=load_balancer['LoadBalancerArn']
)
result = [
attribute
for attribute in filter(
lambda x: x['Key'] == "routing.http.drop_invalid_header_fields.enabled"
and x['Value'] == "true",
response['Attributes'],
)
]
if result: compliant_resource.append(load_balancer['LoadBalancerArn'])
else: non_compliant_resources.append(load_balancer['LoadBalancerArn'])
return RuleCheckResult(
passed=False,
compliant_resources=[],
non_compliant_resources=[],
passed=not non_compliant_resources,
compliant_resources=compliant_resource,
non_compliant_resources=non_compliant_resources,
)
def alb_waf_enabled():
load_balancers = client.describe_load_balancers()
compliant_resource = []
non_compliant_resources = []
for load_balancer in load_balancers['LoadBalancers']:
response = wafv2_client.get_web_acl_for_resource(
ResourceArn=load_balancer['LoadBalancerArn']
)
if 'WebACL' in response: compliant_resource.append(load_balancer['LoadBalancerArn'])
else: non_compliant_resources.append(load_balancer['LoadBalancerArn'])
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
passed=not non_compliant_resources,
compliant_resources=compliant_resource,
non_compliant_resources=non_compliant_resources,
)
def elb_cross_zone_load_balancing_enabled():
load_balancers = client.describe_load_balancers()
compliant_resource = []
non_compliant_resources = []
for load_balancer in load_balancers['LoadBalancers']:
response = client.describe_load_balancer_attributes(
LoadBalancerArn=load_balancer['LoadBalancerArn']
)
result = [
attribute
for attribute in filter(
lambda x: x['Key'] == "load_balancing.cross_zone.enabled"
and x['Value'] == "true",
response['Attributes'],
)
]
if result: compliant_resource.append(load_balancer['LoadBalancerArn'])
else: non_compliant_resources.append(load_balancer['LoadBalancerArn'])
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
passed=not non_compliant_resources,
compliant_resources=compliant_resource,
non_compliant_resources=non_compliant_resources,
)
def elb_deletion_protection_enabled():
load_balancers = client.describe_load_balancers()
compliant_resource = []
non_compliant_resources = []
for load_balancer in load_balancers['LoadBalancers']:
response = client.describe_load_balancer_attributes(
LoadBalancerArn=load_balancer['LoadBalancerArn']
)
result = [
attribute
for attribute in filter(
lambda x: x['Key'] == "deletion_protection.enabled"
and x['Value'] == "true",
response['Attributes'],
)
]
if result: compliant_resource.append(load_balancer['LoadBalancerArn'])
else: non_compliant_resources.append(load_balancer['LoadBalancerArn'])
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
passed=not non_compliant_resources,
compliant_resources=compliant_resource,
non_compliant_resources=non_compliant_resources,
)
def elb_logging_enabled():
load_balancers = client.describe_load_balancers()
compliant_resource = []
non_compliant_resources = []
for load_balancer in load_balancers['LoadBalancers']:
response = client.describe_load_balancer_attributes(
LoadBalancerArn=load_balancer['LoadBalancerArn']
)
result = [
attribute
for attribute in filter(
lambda x: x['Key'] == "connection_logs.s3.enabled"
and x['Value'] == "true",
response['Attributes'],
)
]
if result: compliant_resource.append(load_balancer['LoadBalancerArn'])
else: non_compliant_resources.append(load_balancer['LoadBalancerArn'])
return RuleCheckResult(
passed=False, compliant_resources=[], non_compliant_resources=[]
passed=not non_compliant_resources,
compliant_resources=compliant_resource,
non_compliant_resources=non_compliant_resources,
)