ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(aws): Convert ELB and ELBv2 attributes to dictionaries (#4575)

Browse Source 
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Rubén De la Torre Vico
2024-08-05 17:14:19 +02:00
committed by GitHub
parent d4b6fa27e2
commit fad0b8995a
22 changed files with 363 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/elb/elb_insecure_ssl_ciphers/elb_insecure_ssl_ciphers.py
+2 -2
View File
@@ -8,11 +8,11 @@ class elb_insecure_ssl_ciphers(Check):
secure_ssl_policies = [
"ELBSecurityPolicy-TLS-1-2-2017-01",
]
for lb in elb_client.loadbalancers:
for lb_arn, lb in elb_client.loadbalancers.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "PASS"
report.status_extended = (
prowler/providers/aws/services/elb/elb_internet_facing/elb_internet_facing.py
+2 -2
View File
@@ -5,11 +5,11 @@ from prowler.providers.aws.services.elb.elb_client import elb_client
class elb_internet_facing(Check):
def execute(self):
findings = []
for lb in elb_client.loadbalancers:
for lb_arn, lb in elb_client.loadbalancers.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "PASS"
report.status_extended = f"ELB {lb.name} is not internet facing."
prowler/providers/aws/services/elb/elb_logging_enabled/elb_logging_enabled.py
+2 -2
View File
@@ -5,11 +5,11 @@ from prowler.providers.aws.services.elb.elb_client import elb_client
class elb_logging_enabled(Check):
def execute(self):
findings = []
for lb in elb_client.loadbalancers:
for lb_arn, lb in elb_client.loadbalancers.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "FAIL"
report.status_extended = (
prowler/providers/aws/services/elb/elb_service.py
+32 -32
View File
@@ -7,17 +7,18 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered
from prowler.providers.aws.lib.service.service import AWSService
################### ELB
class ELB(AWSService):
def __init__(self, provider):
# Call AWSService's __init__
super().__init__(__class__.__name__, provider)
self.loadbalancers = []
self.__threading_call__(self.__describe_load_balancers__)
self.__threading_call__(self.__describe_load_balancer_attributes__)
self.__describe_tags__()
self.loadbalancers = {}
self.__threading_call__(self._describe_load_balancers)
self.__threading_call__(
self._describe_load_balancer_attributes, self.loadbalancers.values()
)
self.__threading_call__(self._describe_tags, self.loadbalancers.values())
def __describe_load_balancers__(self, regional_client):
def _describe_load_balancers(self, regional_client):
logger.info("ELB - Describing load balancers...")
try:
describe_elb_paginator = regional_client.get_paginator(
@@ -37,47 +38,47 @@ class ELB(AWSService):
policies=listener["PolicyNames"],
)
)
self.loadbalancers.append(
LoadBalancer(
name=elb["LoadBalancerName"],
arn=arn,
dns=elb["DNSName"],
region=regional_client.region,
scheme=elb["Scheme"],
listeners=listeners,
)
)
self.loadbalancers[arn] = LoadBalancer(
name=elb["LoadBalancerName"],
dns=elb["DNSName"],
region=regional_client.region,
scheme=elb["Scheme"],
listeners=listeners,
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def __describe_load_balancer_attributes__(self, regional_client):
def _describe_load_balancer_attributes(self, load_balancer):
logger.info("ELB - Describing attributes...")
try:
for lb in self.loadbalancers:
if lb.region == regional_client.region:
attributes = regional_client.describe_load_balancer_attributes(
LoadBalancerName=lb.name
)["LoadBalancerAttributes"]
if "AccessLog" in attributes:
lb.access_logs = attributes["AccessLog"]["Enabled"]
regional_client = self.regional_clients[load_balancer.region]
attributes = regional_client.describe_load_balancer_attributes(
LoadBalancerName=load_balancer.name
)["LoadBalancerAttributes"]
load_balancer.access_logs = attributes.get("AccessLog", {}).get(
"Enabled", False
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def __describe_tags__(self):
def _describe_tags(self, load_balancer):
logger.info("ELB - List Tags...")
try:
for lb in self.loadbalancers:
regional_client = self.regional_clients[lb.region]
response = regional_client.describe_tags(LoadBalancerNames=[lb.name])[
"TagDescriptions"
][0]
lb.tags = response.get("Tags")
regional_client = self.regional_clients[load_balancer.region]
tags = regional_client.describe_tags(
LoadBalancerNames=[load_balancer.name]
)["TagDescriptions"][0].get("Tags", [])
load_balancer.tags = tags
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -92,7 +93,6 @@ class Listener(BaseModel):
class LoadBalancer(BaseModel):
name: str
dns: str
arn: str
region: str
scheme: str
access_logs: Optional[bool]
prowler/providers/aws/services/elb/elb_ssl_listeners/elb_ssl_listeners.py
+2 -2
View File
@@ -6,11 +6,11 @@ class elb_ssl_listeners(Check):
def execute(self):
findings = []
secure_protocols = ["SSL", "HTTPS"]
for lb in elb_client.loadbalancers:
for lb_arn, lb in elb_client.loadbalancers.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "PASS"
report.status_extended = f"ELB {lb.name} has HTTPS listeners only."
prowler/providers/aws/services/elbv2/elbv2_deletion_protection/elbv2_deletion_protection.py
+2 -2
View File
@@ -5,11 +5,11 @@ from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
class elbv2_deletion_protection(Check):
def execute(self):
findings = []
for lb in elbv2_client.loadbalancersv2:
for lb_arn, lb in elbv2_client.loadbalancersv2.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "FAIL"
report.status_extended = (
prowler/providers/aws/services/elbv2/elbv2_desync_mitigation_mode/elbv2_desync_mitigation_mode.py
+2 -2
View File
@@ -5,12 +5,12 @@ from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
class elbv2_desync_mitigation_mode(Check):
def execute(self):
findings = []
for lb in elbv2_client.loadbalancersv2:
for lb_arn, lb in elbv2_client.loadbalancersv2.items():
if lb.type == "application":
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "PASS"
report.status_extended = f"ELBv2 ALB {lb.name} is configured with correct desync mitigation mode."
prowler/providers/aws/services/elbv2/elbv2_insecure_ssl_ciphers/elbv2_insecure_ssl_ciphers.py
+3 -3
View File
@@ -17,17 +17,17 @@ class elbv2_insecure_ssl_ciphers(Check):
"ELBSecurityPolicy-TLS13-1-2-Ext1-2021-06",
"ELBSecurityPolicy-TLS13-1-2-Ext2-2021-06",
]
for lb in elbv2_client.loadbalancersv2:
for lb_arn, lb in elbv2_client.loadbalancersv2.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "PASS"
report.status_extended = (
f"ELBv2 {lb.name} does not have insecure SSL protocols or ciphers."
)
for listener in lb.listeners:
for listener in lb.listeners.values():
if (
listener.protocol == "HTTPS"
and listener.ssl_policy not in secure_ssl_policies
prowler/providers/aws/services/elbv2/elbv2_internet_facing/elbv2_internet_facing.py
+2 -2
View File
@@ -5,11 +5,11 @@ from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
class elbv2_internet_facing(Check):
def execute(self):
findings = []
for lb in elbv2_client.loadbalancersv2:
for lb_arn, lb in elbv2_client.loadbalancersv2.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "PASS"
report.status_extended = f"ELBv2 ALB {lb.name} is not internet facing."
prowler/providers/aws/services/elbv2/elbv2_listeners_underneath/elbv2_listeners_underneath.py
+2 -2
View File
@@ -5,11 +5,11 @@ from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
class elbv2_listeners_underneath(Check):
def execute(self):
findings = []
for lb in elbv2_client.loadbalancersv2:
for lb_arn, lb in elbv2_client.loadbalancersv2.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "PASS"
report.status_extended = f"ELBv2 {lb.name} has listeners underneath."
prowler/providers/aws/services/elbv2/elbv2_logging_enabled/elbv2_logging_enabled.py
+2 -2
View File
@@ -5,11 +5,11 @@ from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
class elbv2_logging_enabled(Check):
def execute(self):
findings = []
for lb in elbv2_client.loadbalancersv2:
for lb_arn, lb in elbv2_client.loadbalancersv2.items():
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "FAIL"
report.status_extended = (
prowler/providers/aws/services/elbv2/elbv2_service.py
+113 -147
View File
@@ -1,4 +1,4 @@
from typing import Optional
from typing import Dict, Optional
from botocore.client import ClientError
from pydantic import BaseModel
@@ -8,20 +8,27 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered
from prowler.providers.aws.lib.service.service import AWSService
################### ELBv2
class ELBv2(AWSService):
def __init__(self, provider):
# Call AWSService's __init__
super().__init__(__class__.__name__, provider)
self.loadbalancersv2 = []
self.__threading_call__(self.__describe_load_balancers__)
self.listeners = []
self.__threading_call__(self.__describe_listeners__)
self.__threading_call__(self.__describe_load_balancer_attributes__)
self.__threading_call__(self.__describe_rules__)
self.__describe_tags__()
self.loadbalancersv2 = {}
self.__threading_call__(self._describe_load_balancers)
self.__threading_call__(self._describe_listeners, self.loadbalancersv2.items())
self.__threading_call__(
self._describe_load_balancer_attributes, self.loadbalancersv2.items()
)
self.__threading_call__(
self._describe_rules,
[
(listener_arn, listener)
for lb in self.loadbalancersv2.values()
for listener_arn, listener in lb.listeners.items()
],
)
self.__threading_call__(self._describe_tags, self.loadbalancersv2.items())
def __describe_load_balancers__(self, regional_client):
def _describe_load_balancers(self, regional_client):
logger.info("ELBv2 - Describing load balancers...")
try:
describe_elbv2_paginator = regional_client.get_paginator(
@@ -34,173 +41,134 @@ class ELBv2(AWSService):
elbv2["LoadBalancerArn"], self.audit_resources
)
):
lb = LoadBalancerv2(
self.loadbalancersv2[elbv2["LoadBalancerArn"]] = LoadBalancerv2(
name=elbv2["LoadBalancerName"],
region=regional_client.region,
arn=elbv2["LoadBalancerArn"],
type=elbv2["Type"],
listeners=[],
dns=elbv2.get("DNSName", None),
scheme=elbv2.get("Scheme", None),
)
if "DNSName" in elbv2:
lb.dns = elbv2["DNSName"]
if "Scheme" in elbv2:
lb.scheme = elbv2["Scheme"]
self.loadbalancersv2.append(lb)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def __describe_listeners__(self, regional_client):
def _describe_listeners(self, load_balancer):
logger.info("ELBv2 - Describing listeners...")
try:
for lb in self.loadbalancersv2:
try:
if lb.region == regional_client.region:
describe_elbv2_paginator = regional_client.get_paginator(
"describe_listeners"
)
for page in describe_elbv2_paginator.paginate(
LoadBalancerArn=lb.arn
):
for listener in page["Listeners"]:
port = 0
if "Port" in listener:
port = listener["Port"]
# load_balancer is a tuple with the LoadBalancerArn and the LoadBalancer object
regional_client = self.regional_clients[load_balancer[1].region]
listener_obj = Listenerv2(
region=regional_client.region,
arn=listener["ListenerArn"],
port=port,
ssl_policy=listener.get("SslPolicy"),
rules=[],
)
if "Protocol" in listener:
listener_obj.protocol = listener["Protocol"]
describe_elbv2_paginator = regional_client.get_paginator(
"describe_listeners"
)
lb.listeners.append(listener_obj)
except ClientError as error:
if error.response["Error"]["Code"] == "LoadBalancerNotFound":
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
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}"
for page in describe_elbv2_paginator.paginate(
LoadBalancerArn=load_balancer[0]
):
for listener in page["Listeners"]:
load_balancer[1].listeners[listener["ListenerArn"]] = Listenerv2(
region=regional_client.region,
port=listener.get("Port", 0),
ssl_policy=listener.get("SslPolicy", ""),
protocol=listener.get("Protocol", ""),
)
except ClientError as error:
if error.response["Error"]["Code"] == "LoadBalancerNotFound":
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
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}"
)
def __describe_load_balancer_attributes__(self, regional_client):
def _describe_load_balancer_attributes(self, load_balancer):
logger.info("ELBv2 - Describing attributes...")
try:
for lb in self.loadbalancersv2:
try:
if lb.region == regional_client.region:
for (
attribute
) in regional_client.describe_load_balancer_attributes(
LoadBalancerArn=lb.arn
)[
"Attributes"
]:
if (
attribute["Key"]
== "routing.http.desync_mitigation_mode"
):
lb.desync_mitigation_mode = attribute["Value"]
if attribute["Key"] == "deletion_protection.enabled":
lb.deletion_protection = attribute["Value"]
if attribute["Key"] == "access_logs.s3.enabled":
lb.access_logs = attribute["Value"]
if (
attribute["Key"]
== "routing.http.drop_invalid_header_fields.enabled"
):
lb.drop_invalid_header_fields = attribute["Value"]
regional_client = self.regional_clients[load_balancer[1].region]
except ClientError as error:
if error.response["Error"]["Code"] == "LoadBalancerNotFound":
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
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}"
)
for attribute in regional_client.describe_load_balancer_attributes(
LoadBalancerArn=load_balancer[0]
)["Attributes"]:
if attribute["Key"] == "routing.http.desync_mitigation_mode":
load_balancer[1].desync_mitigation_mode = attribute["Value"]
if attribute["Key"] == "deletion_protection.enabled":
load_balancer[1].deletion_protection = attribute["Value"]
if attribute["Key"] == "access_logs.s3.enabled":
load_balancer[1].access_logs = attribute["Value"]
if (
attribute["Key"]
== "routing.http.drop_invalid_header_fields.enabled"
):
load_balancer[1].drop_invalid_header_fields = attribute["Value"]
except ClientError as error:
if error.response["Error"]["Code"] == "LoadBalancerNotFound":
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
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}"
)
def __describe_rules__(self, regional_client):
def _describe_rules(self, listener):
logger.info("ELBv2 - Describing Rules...")
try:
for lb in self.loadbalancersv2:
if lb.region == regional_client.region:
for listener in lb.listeners:
try:
for rule in regional_client.describe_rules(
ListenerArn=listener.arn
)["Rules"]:
listener.rules.append(
ListenerRule(
arn=rule["RuleArn"],
actions=rule["Actions"],
conditions=rule["Conditions"],
)
)
except ClientError as error:
if error.response["Error"]["Code"] == "ListenerNotFound":
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
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}"
)
# listener is a tuple with the ListenerArn and the Listener object
regional_client = self.regional_clients[listener[1].region]
for rule in regional_client.describe_rules(ListenerArn=listener[0])[
"Rules"
]:
listener[1].rules.append(
ListenerRule(
arn=rule["RuleArn"],
actions=rule["Actions"],
conditions=rule["Conditions"],
)
)
except ClientError as error:
if error.response["Error"]["Code"] == "ListenerNotFound":
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
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}"
)
def __describe_tags__(self):
def _describe_tags(self, load_balancer):
logger.info("ELBv2 - List Tags...")
try:
for lb in self.loadbalancersv2:
try:
regional_client = self.regional_clients[lb.region]
response = regional_client.describe_tags(ResourceArns=[lb.arn])[
"TagDescriptions"
][0]
lb.tags = response.get("Tags")
except ClientError as error:
if error.response["Error"]["Code"] == "LoadBalancerNotFound":
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
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}"
)
regional_client = self.regional_clients[load_balancer[1].region]
load_balancer[1].tags = regional_client.describe_tags(
ResourceArns=[load_balancer[0]]
)["TagDescriptions"][0].get("Tags", [])
except ClientError as error:
if error.response["Error"]["Code"] == "LoadBalancerNotFound":
logger.warning(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
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}"
@@ -214,17 +182,15 @@ class ListenerRule(BaseModel):
class Listenerv2(BaseModel):
arn: str
region: str
port: int
protocol: Optional[str]
ssl_policy: Optional[str]
rules: list[ListenerRule]
protocol: str
ssl_policy: str
rules: list[ListenerRule] = []
class LoadBalancerv2(BaseModel):
name: str
arn: str
region: str
type: str
access_logs: Optional[str]
@@ -232,6 +198,6 @@ class LoadBalancerv2(BaseModel):
deletion_protection: Optional[str]
dns: Optional[str]
drop_invalid_header_fields: Optional[str]
listeners: list[Listenerv2]
listeners: Dict[str, Listenerv2] = {}
scheme: Optional[str]
tags: Optional[list] = []
prowler/providers/aws/services/elbv2/elbv2_ssl_listeners/elbv2_ssl_listeners.py
+3 -3
View File
@@ -5,18 +5,18 @@ from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
class elbv2_ssl_listeners(Check):
def execute(self):
findings = []
for lb in elbv2_client.loadbalancersv2:
for lb_arn, lb in elbv2_client.loadbalancersv2.items():
if lb.type == "application":
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "PASS"
report.status_extended = (
f"ELBv2 ALB {lb.name} has HTTPS listeners only."
)
for listener in lb.listeners:
for listener in lb.listeners.values():
if listener.protocol == "HTTP":
report.status = "FAIL"
report.status_extended = (
prowler/providers/aws/services/elbv2/elbv2_waf_acl_attached/elbv2_waf_acl_attached.py
+4 -4
View File
@@ -7,23 +7,23 @@ from prowler.providers.aws.services.wafv2.wafv2_client import wafv2_client
class elbv2_waf_acl_attached(Check):
def execute(self):
findings = []
for lb in elbv2_client.loadbalancersv2:
for lb_arn, lb in elbv2_client.loadbalancersv2.items():
if lb.type == "application":
report = Check_Report_AWS(self.metadata())
report.region = lb.region
report.resource_id = lb.name
report.resource_arn = lb.arn
report.resource_arn = lb_arn
report.resource_tags = lb.tags
report.status = "FAIL"
report.status_extended = (
f"ELBv2 ALB {lb.name} is not protected by WAF Web ACL."
)
for acl in wafv2_client.web_acls:
if lb.arn in acl.albs:
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:
if lb.arn in acl.albs:
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}."
prowler/providers/aws/services/shield/shield_advanced_protection_in_classic_load_balancers/shield_advanced_protection_in_classic_load_balancers.py
+3 -3
View File
@@ -7,11 +7,11 @@ class shield_advanced_protection_in_classic_load_balancers(Check):
def execute(self):
findings = []
if shield_client.enabled:
for elb in elb_client.loadbalancers:
for elb_arn, elb in elb_client.loadbalancers.items():
report = Check_Report_AWS(self.metadata())
report.region = shield_client.region
report.resource_id = elb.name
report.resource_arn = elb.arn
report.resource_arn = elb_arn
report.resource_tags = elb.tags
report.status = "FAIL"
report.status_extended = (
@@ -19,7 +19,7 @@ class shield_advanced_protection_in_classic_load_balancers(Check):
)
for protection in shield_client.protections.values():
if elb.arn == protection.resource_arn:
if elb_arn == protection.resource_arn:
report.status = "PASS"
report.status_extended = (
f"ELB {elb.name} is protected by AWS Shield Advanced."
prowler/providers/aws/services/shield/shield_advanced_protection_in_internet_facing_load_balancers/shield_advanced_protection_in_internet_facing_load_balancers.py
+3 -3
View File
@@ -7,18 +7,18 @@ class shield_advanced_protection_in_internet_facing_load_balancers(Check):
def execute(self):
findings = []
if shield_client.enabled:
for elbv2 in elbv2_client.loadbalancersv2:
for elbv2_arn, elbv2 in elbv2_client.loadbalancersv2.items():
if elbv2.type == "application" and elbv2.scheme == "internet-facing":
report = Check_Report_AWS(self.metadata())
report.region = shield_client.region
report.resource_id = elbv2.name
report.resource_arn = elbv2.arn
report.resource_arn = elbv2_arn
report.resource_tags = elbv2.tags
report.status = "FAIL"
report.status_extended = f"ELBv2 ALB {elbv2.name} is not protected by AWS Shield Advanced."
for protection in shield_client.protections.values():
if elbv2.arn == protection.resource_arn:
if elbv2_arn == protection.resource_arn:
report.status = "PASS"
report.status_extended = f"ELBv2 ALB {elbv2.name} is protected by AWS Shield Advanced."
break
tests/providers/aws/services/elb/elb_insecure_ssl_ciphers/elb_insecure_ssl_ciphers_test.py
+2
View File
@@ -96,6 +96,7 @@ class Test_elb_insecure_ssl_ciphers:
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == elb_arn
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == []
@mock_aws
def test_elb_with_HTTPS_listener(self):
@@ -146,3 +147,4 @@ class Test_elb_insecure_ssl_ciphers:
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == elb_arn
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == []
tests/providers/aws/services/elb/elb_internet_facing/elb_internet_facing_test.py
+2
View File
@@ -89,6 +89,7 @@ class Test_elb_request_smugling:
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == elb_arn
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == []
@mock_aws
def test_elb_with_deletion_protection(self):
@@ -139,3 +140,4 @@ class Test_elb_request_smugling:
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == elb_arn
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_tags == []
tests/providers/aws/services/elb/elb_logging_enabled/elb_logging_enabled_test.py
+4
View File
@@ -91,6 +91,8 @@ class Test_elb_logging_enabled:
)
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == elb_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
@mock_aws
def test_elb_with_deletion_protection(self):
@@ -152,3 +154,5 @@ class Test_elb_logging_enabled:
)
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == elb_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
tests/providers/aws/services/elb/elb_service_test.py
+59 -18
View File
@@ -37,7 +37,7 @@ class Test_ELB_Service:
# Test ELB Describe Load Balancers
@mock_aws
def test__describe_load_balancers__(self):
def test_describe_load_balancers(self):
elb = client("elb", region_name=AWS_REGION_US_EAST_1)
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
@@ -45,7 +45,7 @@ class Test_ELB_Service:
GroupName="sg01", Description="Test security group sg01"
)
elb.create_load_balancer(
dns_name = elb.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[
{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080},
@@ -54,22 +54,25 @@ class Test_ELB_Service:
AvailabilityZones=[f"{AWS_REGION_US_EAST_1}a"],
Scheme="internal",
SecurityGroups=[security_group.id],
)
)["DNSName"]
elb_arn = f"arn:aws:elasticloadbalancing:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:loadbalancer/my-lb"
# ELB client for this test class
aws_provider = set_mocked_aws_provider()
elb = ELB(aws_provider)
assert len(elb.loadbalancers) == 1
assert elb.loadbalancers[0].name == "my-lb"
assert elb.loadbalancers[0].region == AWS_REGION_US_EAST_1
assert elb.loadbalancers[0].scheme == "internal"
assert (
elb.loadbalancers[0].arn
== f"arn:aws:elasticloadbalancing:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:loadbalancer/my-lb"
)
assert elb.loadbalancers[elb_arn].name == "my-lb"
assert elb.loadbalancers[elb_arn].region == AWS_REGION_US_EAST_1
assert elb.loadbalancers[elb_arn].scheme == "internal"
assert elb.loadbalancers[elb_arn].dns == dns_name
assert len(elb.loadbalancers[elb_arn].listeners) == 2
assert elb.loadbalancers[elb_arn].listeners[0].protocol == "TCP"
assert elb.loadbalancers[elb_arn].listeners[0].policies == []
assert elb.loadbalancers[elb_arn].listeners[1].protocol == "HTTP"
assert elb.loadbalancers[elb_arn].listeners[1].policies == []
# Test ELB Describe Load Balancers Attributes
@mock_aws
def test__describe_load_balancer_attributes__(self):
def test_describe_load_balancer_attributes(self):
elb = client("elb", region_name=AWS_REGION_US_EAST_1)
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
@@ -99,14 +102,52 @@ class Test_ELB_Service:
}
},
)
elb_arn = f"arn:aws:elasticloadbalancing:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:loadbalancer/my-lb"
# ELB client for this test class
aws_provider = set_mocked_aws_provider()
elb = ELB(aws_provider)
assert elb.loadbalancers[0].name == "my-lb"
assert elb.loadbalancers[0].region == AWS_REGION_US_EAST_1
assert elb.loadbalancers[0].scheme == "internal"
assert elb.loadbalancers[0].access_logs
assert (
elb.loadbalancers[0].arn
== f"arn:aws:elasticloadbalancing:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:loadbalancer/my-lb"
assert elb.loadbalancers[elb_arn].name == "my-lb"
assert elb.loadbalancers[elb_arn].region == AWS_REGION_US_EAST_1
assert elb.loadbalancers[elb_arn].scheme == "internal"
assert elb.loadbalancers[elb_arn].access_logs
# Test ELB Describe Tags
@mock_aws
def test_describe_tags(self):
elb = client("elb", region_name=AWS_REGION_US_EAST_1)
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
security_group = ec2.create_security_group(
GroupName="sg01", Description="Test security group sg01"
)
elb.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[
{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080},
{"Protocol": "http", "LoadBalancerPort": 81, "InstancePort": 9000},
],
AvailabilityZones=[f"{AWS_REGION_US_EAST_1}a"],
Scheme="internal",
SecurityGroups=[security_group.id],
)
elb.add_tags(
LoadBalancerNames=["my-lb"],
Tags=[
{"Key": "key1", "Value": "value1"},
{"Key": "key2", "Value": "value2"},
],
)
elb_arn = f"arn:aws:elasticloadbalancing:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:loadbalancer/my-lb"
# ELB client for this test class
aws_provider = set_mocked_aws_provider()
elb = ELB(aws_provider)
assert elb.loadbalancers[elb_arn].name == "my-lb"
assert elb.loadbalancers[elb_arn].region == AWS_REGION_US_EAST_1
assert elb.loadbalancers[elb_arn].scheme == "internal"
assert len(elb.loadbalancers[elb_arn].tags) == 2
assert elb.loadbalancers[elb_arn].tags[0]["Key"] == "key1"
assert elb.loadbalancers[elb_arn].tags[0]["Value"] == "value1"
assert elb.loadbalancers[elb_arn].tags[1]["Key"] == "key2"
assert elb.loadbalancers[elb_arn].tags[1]["Value"] == "value2"
tests/providers/aws/services/elb/elb_ssl_listeners/elb_ssl_listeners_test.py
+4
View File
@@ -88,6 +88,8 @@ class Test_elb_ssl_listeners:
assert result[0].status_extended == "ELB my-lb has non-encrypted listeners."
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == elb_arn
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION
@mock_aws
def test_elb_with_HTTPS_listener(self):
@@ -132,3 +134,5 @@ class Test_elb_ssl_listeners:
assert result[0].status_extended == "ELB my-lb has HTTPS listeners only."
assert result[0].resource_id == "my-lb"
assert result[0].resource_arn == elb_arn
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION
tests/providers/aws/services/elbv2/elbv2_service_test.py
+113 -20
View File
@@ -45,7 +45,7 @@ class Test_ELBv2_Service:
# Test ELBv2 Describe Load Balancers
@mock_aws
def test__describe_load_balancers__(self):
def test_describe_load_balancers(self):
conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
@@ -76,14 +76,22 @@ class Test_ELBv2_Service:
)
elbv2 = ELBv2(aws_provider)
assert len(elbv2.loadbalancersv2) == 1
assert elbv2.loadbalancersv2[0].name == "my-lb"
assert elbv2.loadbalancersv2[0].region == AWS_REGION_EU_WEST_1
assert elbv2.loadbalancersv2[0].scheme == "internal"
assert elbv2.loadbalancersv2[0].arn == lb["LoadBalancerArn"]
assert lb["LoadBalancerArn"] in elbv2.loadbalancersv2.keys()
assert elbv2.loadbalancersv2[lb["LoadBalancerArn"]].name == "my-lb"
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]].region == AWS_REGION_EU_WEST_1
)
assert elbv2.loadbalancersv2[lb["LoadBalancerArn"]].scheme == "internal"
assert elbv2.loadbalancersv2[lb["LoadBalancerArn"]].type == "application"
assert elbv2.loadbalancersv2[lb["LoadBalancerArn"]].listeners == {}
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]].dns
== "my-lb-1.eu-west-1.elb.amazonaws.com"
)
# Test ELBv2 Describe Listeners
@mock_aws
def test__describe_listeners__(self):
def test_describe_listeners(self):
conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
@@ -109,7 +117,7 @@ class Test_ELBv2_Service:
Scheme="internal",
)["LoadBalancers"][0]
conn.create_listener(
listener_arn = conn.create_listener(
LoadBalancerArn=lb["LoadBalancerArn"],
Protocol="HTTP",
Port=443,
@@ -123,19 +131,38 @@ class Test_ELBv2_Service:
},
}
],
)
)["Listeners"][0]["ListenerArn"]
# ELBv2 client for this test class
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
elbv2 = ELBv2(aws_provider)
assert len(elbv2.loadbalancersv2[0].listeners) == 1
assert elbv2.loadbalancersv2[0].listeners[0].protocol == "HTTP"
assert elbv2.loadbalancersv2[0].listeners[0].port == 443
assert len(elbv2.loadbalancersv2[lb["LoadBalancerArn"]].listeners) == 1
assert listener_arn in elbv2.loadbalancersv2[lb["LoadBalancerArn"]].listeners
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]].listeners[listener_arn].region
== AWS_REGION_EU_WEST_1
)
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]]
.listeners[listener_arn]
.protocol
== "HTTP"
)
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]].listeners[listener_arn].port
== 443
)
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]]
.listeners[listener_arn]
.ssl_policy
== "ELBSecurityPolicy-2016-08"
)
# Test ELBv2 Describe Load Balancers Attributes
@mock_aws
def test__describe_load_balancer_attributes__(self):
def test_describe_load_balancer_attributes(self):
conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
@@ -179,14 +206,22 @@ class Test_ELBv2_Service:
)
elbv2 = ELBv2(aws_provider)
assert len(elbv2.loadbalancersv2) == 1
assert elbv2.loadbalancersv2[0].desync_mitigation_mode == "defensive"
assert elbv2.loadbalancersv2[0].access_logs == "true"
assert elbv2.loadbalancersv2[0].deletion_protection == "true"
assert elbv2.loadbalancersv2[0].drop_invalid_header_fields == "false"
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]].desync_mitigation_mode
== "defensive"
)
assert elbv2.loadbalancersv2[lb["LoadBalancerArn"]].access_logs == "true"
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]].deletion_protection == "true"
)
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]].drop_invalid_header_fields
== "false"
)
# Test ELBv2 Describe Load Balancers Attributes
@mock_aws
def test__describe_rules__(self):
def test_describe_rules(self):
conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
@@ -222,15 +257,73 @@ class Test_ELBv2_Service:
},
}
]
conn.create_listener(
listener_arn = conn.create_listener(
LoadBalancerArn=lb["LoadBalancerArn"],
Protocol="HTTP",
DefaultActions=actions,
)
)["Listeners"][0]["ListenerArn"]
# ELBv2 client for this test class
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
elbv2 = ELBv2(aws_provider)
assert len(elbv2.loadbalancersv2) == 1
assert elbv2.loadbalancersv2[0].listeners[0].rules[0].actions == actions
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]]
.listeners[listener_arn]
.rules[0]
.actions
== actions
)
# Test ELBv2 Describe Tags
@mock_aws
def test_describe_tags(self):
conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
security_group = ec2.create_security_group(
GroupName="a-security-group", Description="First One"
)
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
subnet1 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock="172.28.7.192/26",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
)
subnet2 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock="172.28.7.0/26",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZB,
)
lb = conn.create_load_balancer(
Name="my-lb",
Subnets=[subnet1.id, subnet2.id],
SecurityGroups=[security_group.id],
Scheme="internal",
)["LoadBalancers"][0]
conn.add_tags(
ResourceArns=[lb["LoadBalancerArn"]],
Tags=[
{"Key": "Name", "Value": "my-lb"},
{"Key": "Environment", "Value": "dev"},
],
)
# ELBv2 client for this test class
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
elbv2 = ELBv2(aws_provider)
assert len(elbv2.loadbalancersv2) == 1
assert len(elbv2.loadbalancersv2[lb["LoadBalancerArn"]].tags) == 2
assert elbv2.loadbalancersv2[lb["LoadBalancerArn"]].tags[0]["Key"] == "Name"
assert elbv2.loadbalancersv2[lb["LoadBalancerArn"]].tags[0]["Value"] == "my-lb"
assert (
elbv2.loadbalancersv2[lb["LoadBalancerArn"]].tags[1]["Key"] == "Environment"
)
assert elbv2.loadbalancersv2[lb["LoadBalancerArn"]].tags[1]["Value"] == "dev"