ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 20:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(awslambda): add new check awslambda_function_vpc_multi_az (#4816)

Browse Source 
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Rubén De la Torre Vico
2024-09-19 15:30:04 +02:00
committed by GitHub
parent dbc8e140e3
commit 320a2a2c77
11 changed files with 681 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/tutorials/configuration_file.md
+36 -35
View File
@@ -13,52 +13,53 @@ The following list includes all the AWS checks with configurable variables that
| Check Name | Value | Type |
|---------------------------------------------------------------|--------------------------------------------------|-----------------|
| `iam_user_accesskey_unused` | `max_unused_access_keys_days` | Integer |
| `iam_user_console_access_unused` | `max_console_access_days` | Integer |
| `ec2_elastic_ip_shodan` | `shodan_api_key` | String |
| `ec2_securitygroup_with_many_ingress_egress_rules` | `max_security_group_rules` | Integer |
| `ec2_instance_older_than_specific_days` | `max_ec2_instance_age_in_days` | Integer |
| `ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports`| `ec2_sg_high_risk_ports` | List of Integer |
| `vpc_endpoint_connections_trust_boundaries` | `trusted_account_ids` | List of Strings |
| `vpc_endpoint_services_allowed_principals_trust_boundaries` | `trusted_account_ids` | List of Strings |
| `cloudwatch_log_group_retention_policy_specific_days_enabled` | `log_group_retention_days` | Integer |
| `appstream_fleet_session_idle_disconnect_timeout` | `max_idle_disconnect_timeout_in_seconds` | Integer |
| `appstream_fleet_session_disconnect_timeout` | `max_disconnect_timeout_in_seconds` | Integer |
| `appstream_fleet_maximum_session_duration` | `max_session_duration_seconds` | Integer |
| `awslambda_function_using_supported_runtimes` | `obsolete_lambda_runtimes` | Integer |
| `organizations_scp_check_deny_regions` | `organizations_enabled_regions` | List of Strings |
| `organizations_delegated_administrators` | `organizations_trusted_delegated_administrators` | List of Strings |
| `ecr_repositories_scan_vulnerabilities_in_latest_image` | `ecr_repository_vulnerability_minimum_severity` | String |
| `trustedadvisor_premium_support_plan_subscribed` | `verify_premium_support_plans` | Boolean |
| `config_recorder_all_regions_enabled` | `mute_non_default_regions` | Boolean |
| `drs_job_exist` | `mute_non_default_regions` | Boolean |
| `guardduty_is_enabled` | `mute_non_default_regions` | Boolean |
| `securityhub_enabled` | `mute_non_default_regions` | Boolean |
| `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_entropy` | Integer |
| `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_minutes` | Integer |
| `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_actions` | List of Strings |
| `cloudtrail_threat_detection_enumeration` | `threat_detection_enumeration_entropy` | Integer |
| `cloudtrail_threat_detection_enumeration` | `threat_detection_enumeration_minutes` | Integer |
| `cloudtrail_threat_detection_enumeration` | `threat_detection_enumeration_actions` | List of Strings |
| `codebuild_project_no_secrets_in_variables` | `excluded_sensitive_environment_variables` | List of Strings |
| `rds_instance_backup_enabled` | `check_rds_instance_replicas` | Boolean |
| `ec2_securitygroup_allow_ingress_from_internet_to_any_port` | `ec2_allowed_interface_types` | List of Strings |
| `ec2_securitygroup_allow_ingress_from_internet_to_any_port` | `ec2_allowed_instance_owners` | List of Strings |
| `acm_certificates_expiration_check` | `days_to_expire_threshold` | Integer |
| `eks_control_plane_logging_all_types_enabled` | `eks_required_log_types` | List of Strings |
| `eks_cluster_uses_a_supported_version` | `eks_cluster_oldest_version_supported` | String |
| `elbv2_is_in_multiple_az` | `elbv2_min_azs` | Integer |
| `elb_is_in_multiple_az` | `elb_min_azs` | Integer |
| `appstream_fleet_maximum_session_duration` | `max_session_duration_seconds` | Integer |
| `appstream_fleet_session_disconnect_timeout` | `max_disconnect_timeout_in_seconds` | Integer |
| `appstream_fleet_session_idle_disconnect_timeout` | `max_idle_disconnect_timeout_in_seconds` | Integer |
| `autoscaling_find_secrets_ec2_launch_configuration` | `secrets_ignore_patterns` | List of Strings |
| `awslambda_function_no_secrets_in_code` | `secrets_ignore_patterns` | List of Strings |
| `awslambda_function_no_secrets_in_variables` | `secrets_ignore_patterns` | List of Strings |
| `awslambda_function_using_supported_runtimes` | `obsolete_lambda_runtimes` | Integer |
| `awslambda_function_vpc_is_in_multi_azs` | `lambda_min_azs` | Integer |
| `cloudformation_stack_outputs_find_secrets` | `secrets_ignore_patterns` | List of Strings |
| `cloudtrail_threat_detection_enumeration` | `threat_detection_enumeration_actions` | List of Strings |
| `cloudtrail_threat_detection_enumeration` | `threat_detection_enumeration_entropy` | Integer |
| `cloudtrail_threat_detection_enumeration` | `threat_detection_enumeration_minutes` | Integer |
| `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_actions` | List of Strings |
| `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_entropy` | Integer |
| `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_minutes` | Integer |
| `cloudwatch_log_group_no_secrets_in_logs` | `secrets_ignore_patterns` | List of Strings |
| `cloudwatch_log_group_retention_policy_specific_days_enabled` | `log_group_retention_days` | Integer |
| `codebuild_project_no_secrets_in_variables` | `excluded_sensitive_environment_variables` | List of Strings |
| `codebuild_project_no_secrets_in_variables` | `secrets_ignore_patterns` | List of Strings |
| `config_recorder_all_regions_enabled` | `mute_non_default_regions` | Boolean |
| `drs_job_exist` | `mute_non_default_regions` | Boolean |
| `ec2_elastic_ip_shodan` | `shodan_api_key` | String |
| `ec2_instance_older_than_specific_days` | `max_ec2_instance_age_in_days` | Integer |
| `ec2_instance_secrets_user_data` | `secrets_ignore_patterns` | List of Strings |
| `ec2_launch_template_no_secrets` | `secrets_ignore_patterns` | List of Strings |
| `ec2_securitygroup_allow_ingress_from_internet_to_any_port` | `ec2_allowed_instance_owners` | List of Strings |
| `ec2_securitygroup_allow_ingress_from_internet_to_any_port` | `ec2_allowed_interface_types` | List of Strings |
| `ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports`| `ec2_sg_high_risk_ports` | List of Integer |
| `ec2_securitygroup_with_many_ingress_egress_rules` | `max_security_group_rules` | Integer |
| `ecs_task_definitions_no_environment_secrets` | `secrets_ignore_patterns` | List of Strings |
| `ecr_repositories_scan_vulnerabilities_in_latest_image` | `ecr_repository_vulnerability_minimum_severity` | String |
| `eks_cluster_uses_a_supported_version` | `eks_cluster_oldest_version_supported` | String |
| `eks_control_plane_logging_all_types_enabled` | `eks_required_log_types` | List of Strings |
| `elb_is_in_multiple_az` | `elb_min_azs` | Integer |
| `elbv2_is_in_multiple_az` | `elbv2_min_azs` | Integer |
| `guardduty_is_enabled` | `mute_non_default_regions` | Boolean |
| `iam_user_accesskey_unused` | `max_unused_access_keys_days` | Integer |
| `iam_user_console_access_unused` | `max_console_access_days` | Integer |
| `organizations_delegated_administrators` | `organizations_trusted_delegated_administrators` | List of Strings |
| `organizations_scp_check_deny_regions` | `organizations_enabled_regions` | List of Strings |
| `rds_instance_backup_enabled` | `check_rds_instance_replicas` | Boolean |
| `securityhub_enabled` | `mute_non_default_regions` | Boolean |
| `ssm_document_secrets` | `secrets_ignore_patterns` | List of Strings |
| `trustedadvisor_premium_support_plan_subscribed` | `verify_premium_support_plans` | Boolean |
| `vpc_endpoint_connections_trust_boundaries` | `trusted_account_ids` | List of Strings |
| `vpc_endpoint_services_allowed_principals_trust_boundaries` | `trusted_account_ids` | List of Strings |
## Azure
prowler/config/config.yaml
+2
View File
@@ -101,6 +101,8 @@ aws:
"ruby2.5",
"ruby2.7",
]
# aws.awslambda_function_vpc_is_in_multi_azs
lambda_min_azs: 2
# AWS Organizations
# aws.organizations_scp_check_deny_regions
prowler/providers/aws/services/awslambda/awslambda_function_inside_vpc/awslambda_function_inside_vpc.py
+11 -5
View File
@@ -13,14 +13,20 @@ class awslambda_function_inside_vpc(Check):
report.resource_id = function.name
report.resource_arn = function_arn
report.resource_tags = function.tags
report.status = "FAIL"
report.status = "PASS"
report.status_extended = (
f"Lambda function {function.name} is not inside a VPC"
f"Lambda function {function.name} is inside of VPC {function.vpc_id}"
)
if function.vpc_id:
report.status = "PASS"
report.status_extended = f"Lambda function {function.name} is inside of VPC {function.vpc_id}"
if not function.vpc_id:
awslambda_client.set_failed_check(
self.__class__.__name__,
function_arn,
)
report.status = "FAIL"
report.status_extended = (
f"Lambda function {function.name} is not inside a VPC"
)
findings.append(report)
prowler/providers/aws/services/awslambda/awslambda_function_vpc_multi_az/__init__.py
View File
prowler/providers/aws/services/awslambda/awslambda_function_vpc_multi_az/awslambda_function_vpc_multi_az.metadata.json
+30
View File
@@ -0,0 +1,30 @@
{
"Provider": "aws",
"CheckID": "awslambda_function_vpc_multi_az",
"CheckTitle": "Check if AWS Lambda Function VPC is deployed Across Multiple Availability Zones",
"CheckType": [],
"ServiceName": "lambda",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:lambda:region:account-id:function/function-name",
"Severity": "medium",
"ResourceType": "AwsLambdaFunction",
"Description": "This control checks whether an AWS Lambda function connected to a VPC operates in at least the specified number of Availability Zones (AZs). A failure occurs if the function does not operate in the required number of AZs, which by default is two.",
"Risk": "A Lambda function not deployed across multiple AZs increases the risk of a single point of failure, which can result in a complete disruption of the function's operations if an AZ becomes unavailable.",
"RelatedUrl": "https://docs.aws.amazon.com/lambda/latest/operatorguide/networking-vpc.html",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/lambda-controls.html#lambda-5",
"Terraform": ""
},
"Recommendation": {
"Text": "Ensure that your AWS Lambda functions connected to a VPC are distributed across multiple Availability Zones (AZs) to enhance availability and resilience.",
"Url": "https://docs.aws.amazon.com/lambda/latest/dg/configuration-vpc.html"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
prowler/providers/aws/services/awslambda/awslambda_function_vpc_multi_az/awslambda_function_vpc_multi_az.py
+48
View File
@@ -0,0 +1,48 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.awslambda.awslambda_client import awslambda_client
from prowler.providers.aws.services.awslambda.awslambda_function_inside_vpc.awslambda_function_inside_vpc import (
awslambda_function_inside_vpc,
)
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class awslambda_function_vpc_multi_az(Check):
def execute(self) -> list[Check_Report_AWS]:
findings = []
LAMBDA_MIN_AZS = awslambda_client.audit_config.get("lambda_min_azs", 2)
for function_arn, function in awslambda_client.functions.items():
# only proceed if check "awslambda_function_inside_vpc" did not run or did not FAIL to avoid to report that the function is not inside a VPC twice
if not awslambda_client.is_failed_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.status = "FAIL"
report.status_extended = (
f"Lambda function {function.name} is not inside a VPC."
)
if function.vpc_id:
function_availability_zones = {
getattr(
vpc_client.vpc_subnets.get(subnet_id),
"availability_zone",
None,
)
for subnet_id in function.subnet_ids
if subnet_id in vpc_client.vpc_subnets
}
if len(function_availability_zones) >= LAMBDA_MIN_AZS:
report.status = "PASS"
report.status_extended = f"Lambda function {function.name} is inside of VPC {function.vpc_id} that spans in at least {LAMBDA_MIN_AZS} AZs: {', '.join(function_availability_zones)}."
else:
report.status_extended = f"Lambda function {function.name} is inside of VPC {function.vpc_id} that spans only in {len(function_availability_zones)} AZs: {', '.join(function_availability_zones)}. Must span in at least {LAMBDA_MIN_AZS} AZs."
findings.append(report)
return findings
prowler/providers/aws/services/awslambda/awslambda_service.py
+2
View File
@@ -44,6 +44,7 @@ class Lambda(AWSService):
arn=lambda_arn,
security_groups=vpc_config.get("SecurityGroupIds", []),
vpc_id=vpc_config.get("VpcId"),
subnet_ids=set(vpc_config.get("SubnetIds", [])),
region=regional_client.region,
)
if "Runtime" in function:
@@ -202,4 +203,5 @@ class Function(BaseModel):
code: LambdaCode = None
url_config: URLConfig = None
vpc_id: Optional[str]
subnet_ids: Optional[set]
tags: Optional[list] = []
tests/config/config_test.py
+1
View File
@@ -121,6 +121,7 @@ config_aws = {
"ruby2.5",
"ruby2.7",
],
"lambda_min_azs": 2,
"organizations_enabled_regions": [],
"organizations_trusted_delegated_administrators": [],
"ecr_repository_vulnerability_minimum_severity": "MEDIUM",
tests/config/fixtures/config.yaml
+2
View File
@@ -101,6 +101,8 @@ aws:
"ruby2.5",
"ruby2.7",
]
# aws.awslambda_function_vpc_is_in_multi_azs
lambda_min_azs: 2
# AWS Organizations
# aws.organizations_scp_check_deny_regions
tests/providers/aws/services/awslambda/awslambda_function_vpc_multi_az/awslambda_function_vpc_multi_az_test.py
+548
View File
@@ -0,0 +1,548 @@
from json import dumps
from unittest import mock
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_REGION_EU_WEST_1,
AWS_REGION_EU_WEST_1_AZA,
AWS_REGION_EU_WEST_1_AZB,
set_mocked_aws_provider,
)
class Test_awslambda_function_vpc_is_in_multi_azs:
@mock_aws
def test_no_functions(self):
from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.awslambda_client",
new=Lambda(aws_provider),
):
# Test Check
from prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az import (
awslambda_function_vpc_multi_az,
)
check = awslambda_function_vpc_multi_az()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_function_outside_vpc(self):
# Create IAM Role for Lambda Function
iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1)
role_name = "test-role"
assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
role_arn = iam_client.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=dumps(assume_role_policy_document),
)["Role"]["Arn"]
# Create Lambda Function outside VPC
lambda_client = client("lambda", region_name=AWS_REGION_EU_WEST_1)
function_name = "test_function_outside_vpc"
function_arn = lambda_client.create_function(
FunctionName=function_name,
Runtime="python3.8",
Role=role_arn,
Handler="lambda_function.lambda_handler",
Code={"ZipFile": b"file not used"},
)["FunctionArn"]
from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.awslambda_client",
new=Lambda(aws_provider),
):
# Test Check
from prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az import (
awslambda_function_vpc_multi_az,
)
check = awslambda_function_vpc_multi_az()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Lambda function {function_name} is not inside a VPC."
)
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == function_name
assert result[0].resource_arn == function_arn
assert result[0].resource_tags == [{}]
@mock_aws
def test_function_in_vpc_single_az(self):
# Create IAM Role for Lambda Function
iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1)
role_name = "test-role"
assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
role_arn = iam_client.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=dumps(assume_role_policy_document),
)["Role"]["Arn"]
# Create VPC
ec2_client = client("ec2", region_name=AWS_REGION_EU_WEST_1)
vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
# Create Subnet
subnet_id = ec2_client.create_subnet(
VpcId=vpc_id,
CidrBlock="10.0.1.0/24",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
)["Subnet"]["SubnetId"]
# Create Security Group
security_group_id = ec2_client.create_security_group(
GroupName="test-sg", Description="Test SG", VpcId=vpc_id
)["GroupId"]
# Create Lambda Function inside VPC
lambda_client = client("lambda", region_name=AWS_REGION_EU_WEST_1)
function_name = "test_function_in_vpc_single_az"
function = lambda_client.create_function(
FunctionName=function_name,
Runtime="python3.8",
Role=role_arn,
Handler="lambda_function.lambda_handler",
Code={"ZipFile": b"file not used"},
VpcConfig={
"SubnetIds": [subnet_id],
"SecurityGroupIds": [security_group_id],
},
)
function_vpc_id = function["VpcConfig"]["VpcId"]
from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.awslambda_client",
new=Lambda(aws_provider),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.vpc_client",
new=VPC(aws_provider),
):
# Test Check
from prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az import (
awslambda_function_vpc_multi_az,
)
check = awslambda_function_vpc_multi_az()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Lambda function {function_name} is inside of VPC {function_vpc_id} that spans only in 1 AZs: {AWS_REGION_EU_WEST_1_AZA}. Must span in at least 2 AZs."
)
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == function_name
assert result[0].resource_arn == function["FunctionArn"]
assert result[0].resource_tags == [{}]
@mock_aws
def test_function_in_vpc_multiple_az(self):
# Create IAM Role for Lambda Function
iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1)
role_name = "test-role"
assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
role_arn = iam_client.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=dumps(assume_role_policy_document),
)["Role"]["Arn"]
# Create VPC
ec2_client = client("ec2", region_name=AWS_REGION_EU_WEST_1)
vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
# Create Subnets
subnet_id_a = ec2_client.create_subnet(
VpcId=vpc_id,
CidrBlock="10.0.1.0/24",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
)["Subnet"]["SubnetId"]
subnet_id_b = ec2_client.create_subnet(
VpcId=vpc_id,
CidrBlock="10.0.2.0/24",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZB,
)["Subnet"]["SubnetId"]
# Create Security Group
security_group_id = ec2_client.create_security_group(
GroupName="test-sg", Description="Test SG", VpcId=vpc_id
)["GroupId"]
# Create Lambda Function inside VPC
lambda_client = client("lambda", region_name=AWS_REGION_EU_WEST_1)
function_name = "test_function_in_vpc_multiple_az"
function = lambda_client.create_function(
FunctionName=function_name,
Runtime="python3.8",
Role=role_arn,
Handler="lambda_function.lambda_handler",
Code={"ZipFile": b"file not used"},
VpcConfig={
"SubnetIds": [subnet_id_a, subnet_id_b],
"SecurityGroupIds": [security_group_id],
},
)
from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.awslambda_client",
new=Lambda(aws_provider),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.vpc_client",
new=VPC(aws_provider),
):
# Test Check
from prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az import (
awslambda_function_vpc_multi_az,
)
check = awslambda_function_vpc_multi_az()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Lambda function {function_name} is inside of VPC {function['VpcConfig']['VpcId']} that spans in at least 2 AZs: {AWS_REGION_EU_WEST_1_AZB}, {AWS_REGION_EU_WEST_1_AZA}."
) or (
result[0].status_extended
== f"Lambda function {function_name} is inside of VPC {function['VpcConfig']['VpcId']} that spans in at least 2 AZs: {AWS_REGION_EU_WEST_1_AZA}, {AWS_REGION_EU_WEST_1_AZB}."
)
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == function_name
assert result[0].resource_arn == function["FunctionArn"]
assert result[0].resource_tags == [{}]
@mock_aws
def test_function_with_multiple_subnets_in_same_az(self):
# Create IAM Role for Lambda Function
iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1)
role_name = "test-role"
assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
role_arn = iam_client.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=dumps(assume_role_policy_document),
)["Role"]["Arn"]
# Create VPC
ec2_client = client("ec2", region_name=AWS_REGION_EU_WEST_1)
vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
# Create Subnets
subnet_id_a = ec2_client.create_subnet(
VpcId=vpc_id,
CidrBlock="10.0.1.0/24",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
)["Subnet"]["SubnetId"]
subnet_id_b = ec2_client.create_subnet(
VpcId=vpc_id,
CidrBlock="10.0.2.0/24",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
)["Subnet"]["SubnetId"]
# Create Security Group
security_group_id = ec2_client.create_security_group(
GroupName="test-sg", Description="Test SG", VpcId=vpc_id
)["GroupId"]
# Create Lambda Function inside VPC
lambda_client = client("lambda", region_name=AWS_REGION_EU_WEST_1)
function_name = "test_function_in_vpc_multiple_subnets_same_az"
function = lambda_client.create_function(
FunctionName=function_name,
Runtime="python3.8",
Role=role_arn,
Handler="lambda_function.lambda_handler",
Code={"ZipFile": b"file not used"},
VpcConfig={
"SubnetIds": [subnet_id_a, subnet_id_b],
"SecurityGroupIds": [security_group_id],
},
)
from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.awslambda_client",
new=Lambda(aws_provider),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.vpc_client",
new=VPC(aws_provider),
):
# Test Check
from prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az import (
awslambda_function_vpc_multi_az,
)
check = awslambda_function_vpc_multi_az()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Lambda function {function_name} is inside of VPC {function['VpcConfig']['VpcId']} that spans only in 1 AZs: {AWS_REGION_EU_WEST_1_AZA}. Must span in at least 2 AZs."
)
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == function_name
assert result[0].resource_arn == function["FunctionArn"]
assert result[0].resource_tags == [{}]
@mock_aws
def test_function_no_vpc_pass_to_avoid_fail_twice(self):
# Create IAM Role for Lambda Function
iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1)
role_name = "test-role"
assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
role_arn = iam_client.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=dumps(assume_role_policy_document),
)["Role"]["Arn"]
# Create Lambda Function outside VPC
lambda_client = client("lambda", region_name=AWS_REGION_EU_WEST_1)
function_name = "test_function_no_vpc_pass_to_avoid_fail_twice"
lambda_client.create_function(
FunctionName=function_name,
Runtime="python3.8",
Role=role_arn,
Handler="lambda_function.lambda_handler",
Code={"ZipFile": b"file not used"},
)["FunctionArn"]
from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.awslambda_client",
new=Lambda(aws_provider),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.vpc_client",
new=VPC(aws_provider),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_inside_vpc.awslambda_function_inside_vpc.awslambda_client",
new=Lambda(aws_provider),
):
# Test check inside_vpc first
from prowler.providers.aws.services.awslambda.awslambda_function_inside_vpc.awslambda_function_inside_vpc import (
awslambda_function_inside_vpc,
)
check = awslambda_function_inside_vpc()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
# Now test if function is in multiple AZs to ensure it does not fail twice
from prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az import (
awslambda_function_vpc_multi_az,
)
check = awslambda_function_vpc_multi_az()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_resource_filtered(self):
# Create a compliant Lambda Function
# Create IAM Role for Lambda Function
iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1)
role_name = "test-role"
assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
role_arn = iam_client.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=dumps(assume_role_policy_document),
)["Role"]["Arn"]
# Create VPC
ec2_client = client("ec2", region_name=AWS_REGION_EU_WEST_1)
vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
# Create Subnets
subnet_id_a = ec2_client.create_subnet(
VpcId=vpc_id,
CidrBlock="10.0.1.0/24",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
)["Subnet"]["SubnetId"]
subnet_id_b = ec2_client.create_subnet(
VpcId=vpc_id,
CidrBlock="10.0.2.0/24",
AvailabilityZone=AWS_REGION_EU_WEST_1_AZB,
)["Subnet"]["SubnetId"]
# Create Security Group
security_group_id = ec2_client.create_security_group(
GroupName="test-sg", Description="Test SG", VpcId=vpc_id
)["GroupId"]
# Create Lambda Function inside VPC
lambda_client = client("lambda", region_name=AWS_REGION_EU_WEST_1)
function_name = "test_function_resource_filtered"
function = lambda_client.create_function(
FunctionName=function_name,
Runtime="python3.8",
Role=role_arn,
Handler="lambda_function.lambda_handler",
Code={"ZipFile": b"file not used"},
VpcConfig={
"SubnetIds": [subnet_id_a, subnet_id_b],
"SecurityGroupIds": [security_group_id],
},
)
from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
# Filter the resource in the provider, this is like the --resource-arn option
aws_provider._audit_resources = [function["FunctionArn"]]
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.awslambda_client",
new=Lambda(aws_provider),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az.vpc_client",
new=VPC(aws_provider),
):
# Test Check
from prowler.providers.aws.services.awslambda.awslambda_function_vpc_multi_az.awslambda_function_vpc_multi_az import (
awslambda_function_vpc_multi_az,
)
check = awslambda_function_vpc_multi_az()
result = check.execute()
assert len(result) == 1
assert (
result[0].status == "FAIL"
) # This should be a PASS, but the resource is filtered so subnets are filtered and this is a bug
assert (
result[0].status_extended
== f"Lambda function {function_name} is inside of VPC {function['VpcConfig']['VpcId']} that spans only in 0 AZs: . Must span in at least 2 AZs."
)
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == function_name
assert result[0].resource_arn == function["FunctionArn"]
assert result[0].resource_tags == [{}]
tests/providers/aws/services/awslambda/awslambda_service_test.py
+1
View File
@@ -221,6 +221,7 @@ class Test_Lambda_Service:
lambda_arn_1
].url_config.cors_config.allow_origins == ["*"]
assert awslambda.functions[lambda_arn_1].vpc_id == "vpc-123abc"
assert awslambda.functions[lambda_arn_1].subnet_ids == {"subnet-123abc"}
assert awslambda.functions[lambda_arn_1].tags == [{"test": "test"}]