feat(aws): include resource metadata in services from a* to b* (#6504)

Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>
This commit is contained in:
Hugo Pereira Brito
2025-01-16 00:03:37 +01:00
committed by Pepe Fagoaga
parent 69178fd7bd
commit 4c6be5e283
60 changed files with 224 additions and 298 deletions
-2
View File
@@ -420,7 +420,6 @@ class Check_Report:
"""
self.status = ""
self.check_metadata = CheckMetadata.parse_raw(metadata)
if isinstance(resource, dict):
self.resource_metadata = resource
elif isinstance(resource, list):
@@ -436,7 +435,6 @@ class Check_Report:
f"Resource metadata {type(resource)} could not be converted to dict"
)
self.resource_metadata = {}
self.status_extended = ""
self.resource_details = ""
self.resource_tags = getattr(resource, "tags", []) if resource else []
@@ -8,11 +8,9 @@ class accessanalyzer_enabled(Check):
def execute(self):
findings = []
for analyzer in accessanalyzer_client.analyzers:
report = Check_Report_AWS(self.metadata())
report.region = analyzer.region
report.resource_id = analyzer.name
report.resource_arn = analyzer.arn
report.resource_tags = analyzer.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=analyzer
)
if analyzer.status == "ACTIVE":
report.status = "PASS"
report.status_extended = (
@@ -8,14 +8,13 @@ class accessanalyzer_enabled_without_findings(Check):
def execute(self):
findings = []
for analyzer in accessanalyzer_client.analyzers:
report = Check_Report_AWS(self.metadata())
report.region = analyzer.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=analyzer
)
if analyzer.status == "ACTIVE":
report.status = "PASS"
report.status_extended = f"IAM Access Analyzer {analyzer.name} does not have active findings."
report.resource_id = analyzer.name
report.resource_arn = analyzer.arn
report.resource_tags = analyzer.tags
if len(analyzer.findings) != 0:
active_finding_counter = 0
for finding in analyzer.findings:
@@ -25,9 +24,7 @@ class accessanalyzer_enabled_without_findings(Check):
if active_finding_counter > 0:
report.status = "FAIL"
report.status_extended = f"IAM Access Analyzer {analyzer.name} has {active_finding_counter} active findings."
report.resource_id = analyzer.name
report.resource_arn = analyzer.arn
report.resource_tags = analyzer.tags
findings.append(report)
return findings
@@ -6,7 +6,9 @@ from prowler.providers.aws.services.account.account_client import account_client
class account_maintain_current_contact_details(Check):
def execute(self):
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=account_client.contact_base
)
report.region = account_client.region
report.resource_id = account_client.audited_account
report.resource_arn = account_client.audited_account_arn
@@ -8,10 +8,12 @@ class account_maintain_different_contact_details_to_security_billing_and_operati
def execute(self):
findings = []
if account_client.contact_base:
report = Check_Report_AWS(self.metadata())
report.region = account_client.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=account_client.contact_base
)
report.resource_id = account_client.audited_account
report.resource_arn = account_client.audited_account_arn
report.region = account_client.region
if (
len(account_client.contact_phone_numbers)
@@ -6,7 +6,9 @@ from prowler.providers.aws.services.account.account_client import account_client
class account_security_contact_information_is_registered(Check):
def execute(self):
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=account_client.contact_base
)
report.region = account_client.region
report.resource_id = account_client.audited_account
report.resource_arn = account_client.audited_account_arn
@@ -6,7 +6,9 @@ from prowler.providers.aws.services.account.account_client import account_client
class account_security_questions_are_registered_in_the_aws_account(Check):
def execute(self):
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=account_client.contacts_security
)
report.region = account_client.region
report.resource_id = account_client.audited_account
report.resource_arn = account_client.audited_account_arn
@@ -7,17 +7,14 @@ class acm_certificates_expiration_check(Check):
findings = []
for certificate in acm_client.certificates.values():
if certificate.in_use or acm_client.provider.scan_unused_services:
report = Check_Report_AWS(self.metadata())
report.region = certificate.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=certificate
)
if certificate.expiration_days > acm_client.audit_config.get(
"days_to_expire_threshold", 7
):
report.status = "PASS"
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} expires in {certificate.expiration_days} days."
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
else:
report.status = "FAIL"
if certificate.expiration_days < 0:
@@ -26,11 +23,5 @@ class acm_certificates_expiration_check(Check):
else:
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} is about to expire in {certificate.expiration_days} days."
report.check_metadata.Severity = Severity.medium
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
findings.append(report)
return findings
@@ -7,29 +7,18 @@ class acm_certificates_transparency_logs_enabled(Check):
findings = []
for certificate in acm_client.certificates.values():
if certificate.in_use or acm_client.provider.scan_unused_services:
report = Check_Report_AWS(self.metadata())
report.region = certificate.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=certificate
)
if certificate.type == "IMPORTED":
report.status = "PASS"
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} is imported."
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
else:
if not certificate.transparency_logging:
report.status = "FAIL"
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} has Certificate Transparency logging disabled."
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
else:
report.status = "PASS"
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} has Certificate Transparency logging enabled."
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
findings.append(report)
return findings
@@ -7,12 +7,9 @@ class acm_certificates_with_secure_key_algorithms(Check):
findings = []
for certificate in acm_client.certificates.values():
if certificate.in_use or acm_client.provider.scan_unused_services:
report = Check_Report_AWS(self.metadata())
report.region = certificate.region
report.resource_id = certificate.id
report.resource_details = certificate.name
report.resource_arn = certificate.arn
report.resource_tags = certificate.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=certificate
)
report.status = "PASS"
report.status_extended = f"ACM Certificate {certificate.id} for {certificate.name} uses a secure key algorithm ({certificate.key_algorithm})."
@@ -8,11 +8,11 @@ class apigateway_restapi_authorizers_enabled(Check):
def execute(self):
findings = []
for rest_api in apigateway_client.rest_apis:
report = Check_Report_AWS(self.metadata())
report.region = rest_api.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=rest_api
)
report.resource_id = rest_api.name
report.resource_arn = rest_api.arn
report.resource_tags = rest_api.tags
# it there are not authorizers at api level and resources without methods (default case) ->
report.status = "FAIL"
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} does not have an authorizer configured at api level."
@@ -10,11 +10,11 @@ class apigateway_restapi_cache_encrypted(Check):
for rest_api in apigateway_client.rest_apis:
for stage in rest_api.stages:
if stage.cache_enabled:
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=stage
)
report.region = rest_api.region
report.resource_id = rest_api.name
report.resource_arn = stage.arn
report.resource_tags = stage.tags
report.status = "PASS"
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} in stage {stage.name} has cache encryption enabled."
if not stage.cache_data_encrypted:
@@ -9,11 +9,11 @@ class apigateway_restapi_client_certificate_enabled(Check):
findings = []
for rest_api in apigateway_client.rest_apis:
for stage in rest_api.stages:
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=stage
)
report.resource_id = rest_api.name
report.region = rest_api.region
report.resource_arn = stage.arn
report.resource_tags = stage.tags
if stage.client_certificate:
report.status = "PASS"
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} in stage {stage.name} has client certificate enabled."
@@ -9,11 +9,11 @@ class apigateway_restapi_logging_enabled(Check):
findings = []
for rest_api in apigateway_client.rest_apis:
for stage in rest_api.stages:
report = Check_Report_AWS(self.metadata())
report.region = rest_api.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=stage
)
report.resource_id = rest_api.name
report.resource_arn = stage.arn
report.resource_tags = stage.tags
report.region = rest_api.region
if stage.logging:
report.status = "PASS"
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} in stage {stage.name} has logging enabled."
@@ -8,11 +8,11 @@ class apigateway_restapi_public(Check):
def execute(self):
findings = []
for rest_api in apigateway_client.rest_apis:
report = Check_Report_AWS(self.metadata())
report.region = rest_api.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=rest_api
)
report.resource_id = rest_api.name
report.resource_arn = rest_api.arn
report.resource_tags = rest_api.tags
if rest_api.public_endpoint:
report.status = "FAIL"
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} is internet accessible."
@@ -9,11 +9,10 @@ class apigateway_restapi_public_with_authorizer(Check):
findings = []
for rest_api in apigateway_client.rest_apis:
if rest_api.public_endpoint:
report = Check_Report_AWS(self.metadata())
report.region = rest_api.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=rest_api
)
report.resource_id = rest_api.name
report.resource_arn = rest_api.arn
report.resource_tags = rest_api.tags
report.status = "PASS"
report.status_extended = f"API Gateway REST API {rest_api.name} with ID {rest_api.id} has a public endpoint with an authorizer."
@@ -9,11 +9,11 @@ class apigateway_restapi_tracing_enabled(Check):
findings = []
for rest_api in apigateway_client.rest_apis:
for stage in rest_api.stages:
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=stage
)
report.region = rest_api.region
report.resource_id = rest_api.name
report.resource_arn = stage.arn
report.resource_tags = stage.tags
report.status = "FAIL"
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} in stage {stage.name} does not have X-Ray tracing enabled."
if stage.tracing_enabled:
@@ -9,11 +9,11 @@ class apigateway_restapi_waf_acl_attached(Check):
findings = []
for rest_api in apigateway_client.rest_apis:
for stage in rest_api.stages:
report = Check_Report_AWS(self.metadata())
report.region = rest_api.region
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=stage
)
report.resource_id = rest_api.name
report.resource_arn = stage.arn
report.resource_tags = stage.tags
report.region = rest_api.region
if stage.waf:
report.status = "PASS"
report.status_extended = f"API Gateway {rest_api.name} ID {rest_api.id} in stage {stage.name} has {stage.waf} WAF ACL attached."
@@ -8,21 +8,17 @@ class apigatewayv2_api_access_logging_enabled(Check):
def execute(self):
findings = []
for api in apigatewayv2_client.apis:
report = Check_Report_AWS(self.metadata())
report.region = api.region
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=api)
for stage in api.stages:
if stage.logging:
report.status = "PASS"
report.status_extended = f"API Gateway V2 {api.name} ID {api.id} in stage {stage.name} has access logging enabled."
report.resource_id = f"{api.name}-{stage.name}"
report.resource_arn = api.arn
report.resource_tags = api.tags
else:
report.status = "FAIL"
report.status_extended = f"API Gateway V2 {api.name} ID {api.id} in stage {stage.name} has access logging disabled."
report.resource_id = f"{api.name}-{stage.name}"
report.resource_arn = api.arn
report.resource_tags = api.tags
findings.append(report)
return findings
@@ -8,11 +8,8 @@ class apigatewayv2_api_authorizers_enabled(Check):
def execute(self):
findings = []
for api in apigatewayv2_client.apis:
report = Check_Report_AWS(self.metadata())
report.region = api.region
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=api)
report.resource_id = api.name
report.resource_arn = api.arn
report.resource_tags = api.tags
report.status = "FAIL"
report.status_extended = f"API Gateway V2 {api.name} ID {api.id} does not have an authorizer configured."
if api.authorizer:
@@ -10,11 +10,7 @@ class appstream_fleet_default_internet_access_disabled(Check):
"""Execute the appstream_fleet_default_internet_access_disabled check"""
findings = []
for fleet in appstream_client.fleets:
report = Check_Report_AWS(self.metadata())
report.region = fleet.region
report.resource_id = fleet.name
report.resource_arn = fleet.arn
report.resource_tags = fleet.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=fleet)
if fleet.enable_default_internet_access:
report.status = "FAIL"
@@ -15,11 +15,7 @@ class appstream_fleet_maximum_session_duration(Check):
findings = []
for fleet in appstream_client.fleets:
report = Check_Report_AWS(self.metadata())
report.region = fleet.region
report.resource_id = fleet.name
report.resource_arn = fleet.arn
report.resource_tags = fleet.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=fleet)
if fleet.max_user_duration_in_seconds < max_session_duration_seconds:
report.status = "PASS"
@@ -15,11 +15,7 @@ class appstream_fleet_session_disconnect_timeout(Check):
findings = []
for fleet in appstream_client.fleets:
report = Check_Report_AWS(self.metadata())
report.region = fleet.region
report.resource_id = fleet.name
report.resource_arn = fleet.arn
report.resource_tags = fleet.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=fleet)
if fleet.disconnect_timeout_in_seconds <= max_disconnect_timeout_in_seconds:
report.status = "PASS"
@@ -15,11 +15,7 @@ class appstream_fleet_session_idle_disconnect_timeout(Check):
findings = []
for fleet in appstream_client.fleets:
report = Check_Report_AWS(self.metadata())
report.region = fleet.region
report.resource_id = fleet.name
report.resource_arn = fleet.arn
report.resource_tags = fleet.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=fleet)
if (
fleet.idle_disconnect_timeout_in_seconds
@@ -7,11 +7,7 @@ class appsync_field_level_logging_enabled(Check):
findings = []
# Check only GraphQL APIs because boto3 does not have a method to get other types of AppSync APIs (list_apis is not working)
for api in appsync_client.graphql_apis.values():
report = Check_Report_AWS(self.metadata())
report.region = api.region
report.resource_id = api.id
report.resource_arn = api.arn
report.resource_tags = api.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=api)
report.status = "PASS"
report.status_extended = (
f"AppSync API {api.name} has field log level enabled."
@@ -7,11 +7,9 @@ class appsync_graphql_api_no_api_key_authentication(Check):
findings = []
for api in appsync_client.graphql_apis.values():
if api.type == "GRAPHQL":
report = Check_Report_AWS(self.metadata())
report.region = api.region
report.resource_id = api.id
report.resource_arn = api.arn
report.resource_tags = api.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=api
)
report.status = "PASS"
report.status_extended = f"AppSync GraphQL API {api.name} is not using an API KEY for authentication."
if api.authentication_type == "API_KEY":
@@ -13,11 +13,9 @@ class athena_workgroup_encryption(Check):
if (
workgroup.state == "ENABLED" and workgroup.queries
) or athena_client.provider.scan_unused_services:
report = Check_Report_AWS(self.metadata())
report.region = workgroup.region
report.resource_id = workgroup.name
report.resource_arn = workgroup.arn
report.resource_tags = workgroup.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=workgroup
)
if workgroup.encryption_configuration.encrypted:
report.status = "PASS"
@@ -13,11 +13,9 @@ class athena_workgroup_enforce_configuration(Check):
if (
workgroup.state == "ENABLED" and workgroup.queries
) or athena_client.provider.scan_unused_services:
report = Check_Report_AWS(self.metadata())
report.region = workgroup.region
report.resource_id = workgroup.name
report.resource_arn = workgroup.arn
report.resource_tags = workgroup.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=workgroup
)
if workgroup.enforce_workgroup_configuration:
report.status = "PASS"
@@ -21,11 +21,9 @@ class athena_workgroup_logging_enabled(Check):
if (
workgroup.state == "ENABLED" and workgroup.queries
) or athena_client.provider.scan_unused_services:
report = Check_Report_AWS(self.metadata())
report.resource_id = workgroup.name
report.resource_arn = workgroup.arn
report.region = workgroup.region
report.resource_tags = workgroup.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=workgroup
)
report.status = "PASS"
report.status_extended = (
f"Athena WorkGroup {workgroup.name} has CloudWatch logging enabled."
@@ -20,10 +20,9 @@ class autoscaling_find_secrets_ec2_launch_configuration(Check):
configuration_arn,
configuration,
) in autoscaling_client.launch_configurations.items():
report = Check_Report_AWS(self.metadata())
report.region = configuration.region
report.resource_id = configuration.name
report.resource_arn = configuration_arn
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=configuration
)
if configuration.user_data:
user_data = b64decode(configuration.user_data)
@@ -9,11 +9,10 @@ class autoscaling_group_capacity_rebalance_enabled(Check):
findings = []
for group in autoscaling_client.groups:
if group.load_balancers and group.target_groups:
report = Check_Report_AWS(self.metadata())
report.region = group.region
report.resource_id = group.name
report.resource_arn = group.arn
report.resource_tags = group.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=group
)
report.status = "FAIL"
report.status_extended = f"Autoscaling group {group.name} does not have capacity rebalance enabled."
if group.capacity_rebalance:
@@ -9,11 +9,10 @@ class autoscaling_group_elb_health_check_enabled(Check):
findings = []
for group in autoscaling_client.groups:
if group.load_balancers and group.target_groups:
report = Check_Report_AWS(self.metadata())
report.region = group.region
report.resource_id = group.name
report.resource_arn = group.arn
report.resource_tags = group.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=group
)
report.status = "FAIL"
report.status_extended = f"Autoscaling group {group.name} is associated with a load balancer but does not have ELB health checks enabled, instead it has {group.health_check_type} health checks."
if "ELB" in group.health_check_type:
@@ -10,11 +10,10 @@ class autoscaling_group_launch_configuration_no_public_ip(Check):
for group in autoscaling_client.groups:
for lc in autoscaling_client.launch_configurations.values():
if lc.name == group.launch_configuration_name:
report = Check_Report_AWS(self.metadata())
report.region = group.region
report.resource_id = group.name
report.resource_arn = group.arn
report.resource_tags = group.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=group
)
report.status = "PASS"
report.status_extended = f"Autoscaling group {group.name} does not have an associated launch configuration assigning a public IP address."
@@ -12,11 +12,10 @@ class autoscaling_group_launch_configuration_requires_imdsv2(Check):
launch_configuration
) in autoscaling_client.launch_configurations.values():
if launch_configuration.name == group.launch_configuration_name:
report = Check_Report_AWS(self.metadata())
report.region = group.region
report.resource_id = group.name
report.resource_arn = group.arn
report.resource_tags = group.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=group
)
report.status = "FAIL"
report.status_extended = f"Autoscaling group {group.name} has IMDSv2 disabled or not required."
if (
@@ -8,11 +8,8 @@ class autoscaling_group_multiple_az(Check):
def execute(self):
findings = []
for group in autoscaling_client.groups:
report = Check_Report_AWS(self.metadata())
report.region = group.region
report.resource_id = group.name
report.resource_arn = group.arn
report.resource_tags = group.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=group)
report.status = "FAIL"
report.status_extended = (
f"Autoscaling group {group.name} has only one availability zones."
@@ -8,11 +8,8 @@ class autoscaling_group_multiple_instance_types(Check):
def execute(self):
findings = []
for group in autoscaling_client.groups:
report = Check_Report_AWS(self.metadata())
report.region = group.region
report.resource_id = group.name
report.resource_arn = group.arn
report.resource_tags = group.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=group)
report.status = "FAIL"
report.status_extended = f"Autoscaling group {group.name} does not have multiple instance types in multiple Availability Zones."
@@ -8,11 +8,8 @@ class autoscaling_group_using_ec2_launch_template(Check):
def execute(self):
findings = []
for group in autoscaling_client.groups:
report = Check_Report_AWS(self.metadata())
report.region = group.region
report.resource_id = group.name
report.resource_arn = group.arn
report.resource_tags = group.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=group)
report.status = "PASS"
report.status_extended = (
f"Autoscaling group {group.name} is using an EC2 launch template."
@@ -8,11 +8,10 @@ class awslambda_function_inside_vpc(Check):
def execute(self) -> List[Check_Report_AWS]:
findings = []
for function_arn, function in awslambda_client.functions.items():
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function_arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
report.status = "PASS"
report.status_extended = (
f"Lambda function {function.name} is inside of VPC {function.vpc_id}"
@@ -9,11 +9,9 @@ class awslambda_function_invoke_api_operations_cloudtrail_logging_enabled(Check)
def execute(self):
findings = []
for function in awslambda_client.functions.values():
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function.arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
report.status = "FAIL"
report.status_extended = (
@@ -15,11 +15,9 @@ class awslambda_function_no_secrets_in_code(Check):
)
for function, function_code in awslambda_client._get_function_code():
if function_code:
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function.arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
report.status = "PASS"
report.status_extended = (
@@ -13,11 +13,9 @@ class awslambda_function_no_secrets_in_variables(Check):
"secrets_ignore_patterns", []
)
for function in awslambda_client.functions.values():
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function.arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
report.status = "PASS"
report.status_extended = (
@@ -7,11 +7,9 @@ class awslambda_function_not_publicly_accessible(Check):
def execute(self):
findings = []
for function in awslambda_client.functions.values():
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function.arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
report.status = "PASS"
report.status_extended = f"Lambda function {function.name} has a policy resource-based policy not public."
@@ -6,11 +6,10 @@ class awslambda_function_url_cors_policy(Check):
def execute(self):
findings = []
for function in awslambda_client.functions.values():
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function.arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
if function.url_config:
if "*" in function.url_config.cors_config.allow_origins:
report.status = "FAIL"
@@ -7,11 +7,10 @@ class awslambda_function_url_public(Check):
def execute(self):
findings = []
for function in awslambda_client.functions.values():
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function.arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
if function.url_config:
if function.url_config.auth_type == AuthType.AWS_IAM:
report.status = "PASS"
@@ -33,11 +33,9 @@ class awslambda_function_using_supported_runtimes(Check):
findings = []
for function in awslambda_client.functions.values():
if function.runtime:
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function.arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
if function.runtime in awslambda_client.audit_config.get(
"obsolete_lambda_runtimes", default_obsolete_lambda_runtimes
@@ -16,11 +16,10 @@ class awslambda_function_vpc_multi_az(Check):
awslambda_function_inside_vpc.__name__,
function_arn,
):
report = Check_Report_AWS(self.metadata())
report.region = function.region
report.resource_id = function.name
report.resource_arn = function_arn
report.resource_tags = function.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=function
)
report.status = "FAIL"
report.status_extended = (
f"Lambda function {function.name} is not inside a VPC."
@@ -6,21 +6,20 @@ class backup_plans_exist(Check):
def execute(self):
findings = []
if backup_client.backup_plans:
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(),
resource_metadata=backup_client.backup_plans[0],
)
report.status = "PASS"
report.status_extended = f"At least one Backup Plan exists: {backup_client.backup_plans[0].name}."
report.resource_arn = backup_client.backup_plans[0].arn
report.resource_id = backup_client.backup_plans[0].name
report.region = backup_client.backup_plans[0].region
report.resource_tags = backup_client.backup_plans[0].tags
findings.append(report)
elif backup_client.backup_vaults:
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(self.metadata(), resource_metadata=backup_client)
report.status = "FAIL"
report.status_extended = "No Backup Plan exist."
report.resource_arn = backup_client.backup_plan_arn_template
report.resource_id = backup_client.audited_account
report.region = backup_client.region
report.resource_tags = []
findings.append(report)
return findings
@@ -6,11 +6,10 @@ class backup_recovery_point_encrypted(Check):
def execute(self):
findings = []
for recovery_point in backup_client.recovery_points:
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=recovery_point
)
report.region = recovery_point.backup_vault_region
report.resource_id = recovery_point.id
report.resource_arn = recovery_point.arn
report.resource_tags = recovery_point.tags
report.status = "FAIL"
report.status_extended = f"Backup Recovery Point {recovery_point.id} for Backup Vault {recovery_point.backup_vault_name} is not encrypted at rest."
if recovery_point.encrypted:
@@ -7,13 +7,20 @@ class backup_reportplans_exist(Check):
findings = []
# We only check report plans if backup plans exist
if backup_client.backup_plans:
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No Backup Report Plan exist."
report = Check_Report_AWS(
metadata=self.metadata(),
resource_metadata=backup_client.backup_plans[0],
)
report.resource_arn = backup_client.report_plan_arn_template
report.resource_id = backup_client.audited_account
report.region = backup_client.region
report.status = "FAIL"
report.status_extended = "No Backup Report Plan exist."
if backup_client.backup_report_plans:
report = Check_Report_AWS(
metadata=self.metadata(),
resource_metadata=backup_client.backup_report_plans[0],
)
report.status = "PASS"
report.status_extended = f"At least one backup report plan exists: {backup_client.backup_report_plans[0].name}."
report.resource_arn = backup_client.backup_report_plans[0].arn
@@ -7,11 +7,9 @@ class backup_vaults_encrypted(Check):
findings = []
if backup_client.backup_vaults:
for backup_vault in backup_client.backup_vaults:
report = Check_Report_AWS(self.metadata())
report.resource_arn = backup_vault.arn
report.resource_id = backup_vault.name
report.region = backup_vault.region
report.resource_tags = backup_vault.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=backup_vault
)
report.status = "FAIL"
report.status_extended = (
f"Backup Vault {backup_vault.name} is not encrypted at rest."
@@ -6,20 +6,21 @@ class backup_vaults_exist(Check):
def execute(self):
findings = []
if backup_client.backup_vaults is not None:
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No Backup Vault exist."
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=backup_client.backup_vaults
)
report.resource_arn = backup_client.backup_vault_arn_template
report.resource_id = backup_client.audited_account
report.region = backup_client.region
report.resource_tags = []
report.status = "FAIL"
report.status_extended = "No Backup Vault exist."
if backup_client.backup_vaults:
report = Check_Report_AWS(
metadata=self.metadata(),
resource_metadata=backup_client.backup_vaults[0],
)
report.status = "PASS"
report.status_extended = f"At least one backup vault exists: {backup_client.backup_vaults[0].name}."
report.resource_arn = backup_client.backup_vaults[0].arn
report.resource_id = backup_client.backup_vaults[0].name
report.region = backup_client.backup_vaults[0].region
report.resource_tags = backup_client.backup_vaults[0].tags
findings.append(report)
return findings
@@ -8,11 +8,7 @@ class bedrock_agent_guardrail_enabled(Check):
def execute(self):
findings = []
for agent in bedrock_agent_client.agents.values():
report = Check_Report_AWS(self.metadata())
report.region = agent.region
report.resource_id = agent.id
report.resource_arn = agent.arn
report.resource_tags = agent.tags
report = Check_Report_AWS(metadata=self.metadata(), resource_metadata=agent)
report.status = "FAIL"
report.status_extended = f"Bedrock Agent {agent.name} is not using any guardrail to protect agent sessions."
if agent.guardrail_id:
@@ -6,11 +6,9 @@ class bedrock_guardrail_prompt_attack_filter_enabled(Check):
def execute(self):
findings = []
for guardrail in bedrock_client.guardrails.values():
report = Check_Report_AWS(self.metadata())
report.region = guardrail.region
report.resource_id = guardrail.id
report.resource_arn = guardrail.arn
report.resource_tags = guardrail.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=guardrail
)
report.status = "PASS"
report.status_extended = f"Bedrock Guardrail {guardrail.name} is configured to detect and block prompt attacks with a HIGH strength."
if not guardrail.prompt_attack_filter_strength:
@@ -6,11 +6,9 @@ class bedrock_guardrail_sensitive_information_filter_enabled(Check):
def execute(self):
findings = []
for guardrail in bedrock_client.guardrails.values():
report = Check_Report_AWS(self.metadata())
report.region = guardrail.region
report.resource_id = guardrail.id
report.resource_arn = guardrail.arn
report.resource_tags = guardrail.tags
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=guardrail
)
report.status = "PASS"
report.status_extended = f"Bedrock Guardrail {guardrail.name} is blocking or masking sensitive information."
if not guardrail.sensitive_information_filter:
@@ -6,7 +6,9 @@ class bedrock_model_invocation_logging_enabled(Check):
def execute(self):
findings = []
for region, logging in bedrock_client.logging_configurations.items():
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=logging
)
report.region = region
report.resource_id = "model-invocation-logging"
report.resource_arn = (
@@ -11,7 +11,9 @@ class bedrock_model_invocation_logs_encryption_enabled(Check):
if logging.enabled:
s3_encryption = True
cloudwatch_encryption = True
report = Check_Report_AWS(self.metadata())
report = Check_Report_AWS(
metadata=self.metadata(), resource_metadata=logging
)
report.region = region
report.resource_id = "model-invocation-logging"
report.resource_arn = (
@@ -1,16 +1,17 @@
from unittest import mock
from prowler.providers.aws.services.account.account_service import Contact
from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_ACCOUNT_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
AWS_REGION = "us-east-1"
class Test_account_maintain_different_contact_details_to_security_billing_and_operations:
def test_contacts_not_configured_or_equal(self):
account_client = mock.MagicMock
account_client.region = AWS_REGION
aws_mocked_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
account_client = mock.MagicMock()
account_client.region = AWS_REGION_EU_WEST_1
account_client.audited_account = AWS_ACCOUNT_NUMBER
account_client.audited_account_arn = AWS_ACCOUNT_ARN
@@ -26,12 +27,15 @@ class Test_account_maintain_different_contact_details_to_security_billing_and_op
account_client.contact_names = {}
account_client.contact_emails = {}
with mock.patch(
"prowler.providers.aws.services.account.account_service.Account",
new=account_client,
), mock.patch(
"prowler.providers.aws.services.account.account_client.account_client",
new=account_client,
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_mocked_provider,
),
mock.patch(
"prowler.providers.aws.services.account.account_maintain_different_contact_details_to_security_billing_and_operations.account_maintain_different_contact_details_to_security_billing_and_operations.account_client",
new=account_client,
),
):
# Test Check
from prowler.providers.aws.services.account.account_maintain_different_contact_details_to_security_billing_and_operations.account_maintain_different_contact_details_to_security_billing_and_operations import (
@@ -49,13 +53,14 @@ class Test_account_maintain_different_contact_details_to_security_billing_and_op
result[0].status_extended
== "SECURITY, BILLING and OPERATIONS contacts not found or they are not different between each other and between ROOT contact."
)
assert result[0].region == AWS_REGION
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == AWS_ACCOUNT_ARN
def test_contacts_diffent(self):
account_client = mock.MagicMock
account_client.region = AWS_REGION
def test_contacts_different(self):
aws_mocked_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
account_client = mock.MagicMock()
account_client.region = AWS_REGION_EU_WEST_1
account_client.audited_account = AWS_ACCOUNT_NUMBER
account_client.audited_account_arn = AWS_ACCOUNT_ARN
@@ -75,12 +80,15 @@ class Test_account_maintain_different_contact_details_to_security_billing_and_op
"test3@test.com",
}
with mock.patch(
"prowler.providers.aws.services.account.account_service.Account",
new=account_client,
), mock.patch(
"prowler.providers.aws.services.account.account_client.account_client",
new=account_client,
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_mocked_provider,
),
mock.patch(
"prowler.providers.aws.services.account.account_maintain_different_contact_details_to_security_billing_and_operations.account_maintain_different_contact_details_to_security_billing_and_operations.account_client",
new=account_client,
),
):
# Test Check
from prowler.providers.aws.services.account.account_maintain_different_contact_details_to_security_billing_and_operations.account_maintain_different_contact_details_to_security_billing_and_operations import (
@@ -98,23 +106,27 @@ class Test_account_maintain_different_contact_details_to_security_billing_and_op
result[0].status_extended
== "SECURITY, BILLING and OPERATIONS contacts found and they are different between each other and between ROOT contact."
)
assert result[0].region == AWS_REGION
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == AWS_ACCOUNT_ARN
def test_access_denied(self):
account_client = mock.MagicMock
account_client.region = AWS_REGION
aws_mocked_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
account_client = mock.MagicMock()
account_client.region = AWS_REGION_EU_WEST_1
account_client.audited_account = AWS_ACCOUNT_NUMBER
account_client.audited_account_arn = AWS_ACCOUNT_ARN
account_client.contact_base = None
with mock.patch(
"prowler.providers.aws.services.account.account_service.Account",
new=account_client,
), mock.patch(
"prowler.providers.aws.services.account.account_client.account_client",
new=account_client,
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_mocked_provider,
),
mock.patch(
"prowler.providers.aws.services.account.account_maintain_different_contact_details_to_security_billing_and_operations.account_maintain_different_contact_details_to_security_billing_and_operations.account_client",
new=account_client,
),
):
# Test Check
from prowler.providers.aws.services.account.account_maintain_different_contact_details_to_security_billing_and_operations.account_maintain_different_contact_details_to_security_billing_and_operations import (
@@ -10,7 +10,7 @@ AWS_ACCOUNT_NUMBER = "123456789012"
class Test_backup_plans_exist:
def test_no_backup_plans(self):
backup_client = mock.MagicMock
backup_client = mock.MagicMock()
backup_client.audited_account = AWS_ACCOUNT_NUMBER
backup_client.audited_account_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
backup_client.audited_partition = "aws"
@@ -34,7 +34,7 @@ class Test_backup_reportplans_exist:
assert len(result) == 0
def test_no_backup_report_plans(self):
backup_client = mock.MagicMock
backup_client = mock.MagicMock()
backup_client.audited_account = AWS_ACCOUNT_NUMBER
backup_client.audited_account_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
backup_client.region = AWS_REGION
@@ -85,7 +85,7 @@ class Test_backup_reportplans_exist:
assert result[0].region == AWS_REGION
def test_one_backup_report_plan(self):
backup_client = mock.MagicMock
backup_client = mock.MagicMock()
backup_client.audited_account = AWS_ACCOUNT_NUMBER
backup_client.audited_account_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
backup_client.region = AWS_REGION
@@ -8,7 +8,7 @@ AWS_ACCOUNT_NUMBER = "123456789012"
class Test_backup_vaults_exist:
def test_no_backup_vaults(self):
backup_client = mock.MagicMock
backup_client = mock.MagicMock()
backup_client.audited_account = AWS_ACCOUNT_NUMBER
backup_client.audited_account_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
backup_client.region = AWS_REGION
@@ -45,7 +45,7 @@ class Test_backup_vaults_exist:
assert result[0].resource_tags == []
def test_one_backup_vault(self):
backup_client = mock.MagicMock
backup_client = mock.MagicMock()
backup_client.audited_account = AWS_ACCOUNT_NUMBER
backup_client.audited_account_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
backup_client.region = AWS_REGION
@@ -96,7 +96,7 @@ class Test_backup_vaults_exist:
assert result[0].resource_tags == []
def test_access_denied(self):
backup_client = mock.MagicMock
backup_client = mock.MagicMock()
backup_client.audited_account = AWS_ACCOUNT_NUMBER
backup_client.audited_account_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
backup_client.region = AWS_REGION