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

feat(ec2): Ensure both VPN tunnels for an AWS Site-to-Site VPN connection are UP (#4948)

Browse Source
This commit is contained in:
Mario Rodriguez Lopez
2024-09-12 14:26:35 +02:00
committed by GitHub
parent 9f5a909be3
commit c0c59968bf
6 changed files with 318 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/vpc/vpc_service.py
+43
View File
@@ -31,6 +31,8 @@ class VPC(AWSService):
self.vpc_subnets = {}
self.__threading_call__(self._describe_vpc_subnets)
self._describe_network_interfaces()
self.vpn_connections = {}
self.__threading_call__(self._describe_vpn_connections)
def _describe_vpcs(self, regional_client):
logger.info("VPC - Describing VPCs...")
@@ -399,6 +401,35 @@ class VPC(AWSService):
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _describe_vpn_connections(self, regional_client):
try:
describe_vpn_connections = regional_client.describe_vpn_connections()
for vpn_connection in describe_vpn_connections["VpnConnections"]:
arn = f"arn:{self.audited_partition}:ec2:{regional_client.region}:{self.audited_account}:vpn-connection/{vpn_connection['VpnConnectionId']}"
if not self.audit_resources or (
is_resource_filtered(arn, self.audit_resources)
):
tunnels = []
for tunnel in vpn_connection["VgwTelemetry"]:
tunnels.append(
VpnTunnel(
status=tunnel["Status"],
outside_ip_address=tunnel["OutsideIpAddress"],
)
)
self.vpn_connections[arn] = VpnConnection(
id=vpn_connection["VpnConnectionId"],
tunnels=tunnels,
region=regional_client.region,
tags=vpn_connection.get("Tags"),
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class VpcSubnet(BaseModel):
arn: str
@@ -467,3 +498,15 @@ class VpcEndpointService(BaseModel):
allowed_principals: list = []
region: str
tags: Optional[list] = []
class VpnTunnel(BaseModel):
status: str
outside_ip_address: str
class VpnConnection(BaseModel):
id: str
tunnels: list[VpnTunnel]
region: str
tags: Optional[list] = []
prowler/providers/aws/services/vpc/vpc_vpn_connection_tunnels_up/__init__.py
View File
prowler/providers/aws/services/vpc/vpc_vpn_connection_tunnels_up/vpc_vpn_connection_tunnels_up.metadata.json
+34
View File
@@ -0,0 +1,34 @@
{
"Provider": "aws",
"CheckID": "vpc_vpn_connection_tunnels_up",
"CheckTitle": "Both VPN tunnels for an AWS Site-to-Site VPN connection should be up",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "vpc",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:service:region:account-id:vpn-connection/resource-id",
"Severity": "medium",
"ResourceType": "AwsEc2VPNConnection",
"Description": "A VPN tunnel is an encrypted link where data can pass from the customer network to or from AWS within an AWS Site-to-Site VPN connection. Each VPN connection includes two VPN tunnels which you can simultaneously use for high availability. Ensuring that both VPN tunnels are up for a VPN connection is important for confirming a secure and highly available connection between an AWS VPC and your remote network.",
"Risk": "If one or both VPN tunnels are down, it can compromise the security and availability of the connection between your AWS VPC and your remote network. This could result in connectivity issues and potential data exposure or loss during the downtime, affecting business operations and overall network security.",
"RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/vpc-vpn-2-tunnels-up.html",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/ec2-controls.html#ec2-20",
"Terraform": ""
},
"Recommendation": {
"Text": "To modify VPN tunnel options, see Modifying Site-to-Site VPN tunnel options in the AWS Site-to-Site VPN User Guide.",
"Url": "https://docs.aws.amazon.com/vpn/latest/s2svpn/modify-vpn-tunnel-options.html"
}
},
"Categories": [
"redundancy"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
prowler/providers/aws/services/vpc/vpc_vpn_connection_tunnels_up/vpc_vpn_connection_tunnels_up.py
+31
View File
@@ -0,0 +1,31 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class vpc_vpn_connection_tunnels_up(Check):
def execute(self):
findings = []
for vpn_arn, vpn_connection in vpc_client.vpn_connections.items():
report = Check_Report_AWS(self.metadata())
report.region = vpn_connection.region
report.resource_id = vpn_connection.id
report.resource_arn = vpn_arn
report.resource_tags = vpn_connection.tags
if (
vpn_connection.tunnels[0].status != "UP"
or vpn_connection.tunnels[1].status != "UP"
):
report.status = "FAIL"
report.status_extended = (
f"VPN Connection {vpn_connection.id} has at least one tunnel DOWN. "
)
else:
report.status = "PASS"
report.status_extended = (
f"VPN Connection {vpn_connection.id} has both tunnels UP. "
)
findings.append(report)
return findings
tests/providers/aws/services/vpc/vpc_service_test.py
+50 -1
View File
@@ -1,9 +1,11 @@
import json
import botocore
import mock
from boto3 import client, resource
from moto import mock_aws
from prowler.providers.aws.services.vpc.vpc_service import Route
from prowler.providers.aws.services.vpc.vpc_service import VPC, Route
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
@@ -11,6 +13,37 @@ from tests.providers.aws.utils import (
set_mocked_aws_provider,
)
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeVpnConnections":
return {
"VpnConnections": [
{
"VpnConnectionId": "vpn-1234567890abcdef0",
"CustomerGatewayId": "cgw-0123456789abcdef0",
"VpnGatewayId": "vgw-0123456789abcdef0",
"State": "available",
"Type": "ipsec.1",
"VgwTelemetry": [
{
"OutsideIpAddress": "192.168.1.1",
"Status": "UP",
"AcceptedRouteCount": 10,
},
{
"OutsideIpAddress": "192.168.1.2",
"Status": "UP",
"AcceptedRouteCount": 5,
},
],
"Tags": [{"Key": "Name", "Value": "MyVPNConnection"}],
}
]
}
return make_api_call(self, operation_name, kwarg)
class Test_VPC_Service:
@@ -348,3 +381,19 @@ class Test_VPC_Service:
assert vpc.subnets[0].nat_gateway is False
assert vpc.subnets[0].region == AWS_REGION_US_EAST_1
assert vpc.subnets[0].tags is None
# Test VPC Describe VPN Connections
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_describe_vpn_connections(self):
# Generate VPC Client
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
vpc = VPC(aws_provider)
vpn_arn = f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{aws_provider.identity.account}:vpn-connection/vpn-1234567890abcdef0"
assert len(vpc.vpn_connections) == 1
assert vpn_arn in vpc.vpn_connections
vpn_conn = vpc.vpn_connections[vpn_arn]
assert vpn_conn.id == "vpn-1234567890abcdef0"
assert vpn_conn.region == AWS_REGION_US_EAST_1
assert len(vpn_conn.tunnels) == 2
tests/providers/aws/services/vpc/vpc_vpn_connection_tunnels_up/vpc_vpn_connection_tunnels_up_test.py
+160
View File
@@ -0,0 +1,160 @@
from unittest import mock
import botocore
from moto import mock_aws
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeVpnConnections":
return {
"VpnConnections": [
{
"VpnConnectionId": "vpn-1234567890abcdef0",
"CustomerGatewayId": "cgw-0123456789abcdef0",
"VpnGatewayId": "vgw-0123456789abcdef0",
"State": "available",
"Type": "ipsec.1",
"VgwTelemetry": [
{
"OutsideIpAddress": "192.168.1.1",
"Status": "UP",
"AcceptedRouteCount": 10,
},
{
"OutsideIpAddress": "192.168.1.2",
"Status": "UP",
"AcceptedRouteCount": 5,
},
],
"Tags": [{"Key": "Name", "Value": "MyVPNConnection"}],
}
]
}
return make_api_call(self, operation_name, kwarg)
def mock_make_api_call_v2(self, operation_name, kwarg):
if operation_name == "DescribeVpnConnections":
return {
"VpnConnections": [
{
"VpnConnectionId": "vpn-1234567890abcdef0",
"CustomerGatewayId": "cgw-0123456789abcdef0",
"VpnGatewayId": "vgw-0123456789abcdef0",
"State": "available",
"Type": "ipsec.1",
"VgwTelemetry": [
{
"OutsideIpAddress": "192.168.1.1",
"Status": "UP",
"AcceptedRouteCount": 10,
},
{
"OutsideIpAddress": "192.168.1.2",
"Status": "DOWN",
"AcceptedRouteCount": 5,
},
],
"Tags": [{"Key": "Name", "Value": "MyVPNConnection"}],
}
]
}
return make_api_call(self, operation_name, kwarg)
class Test_vpc_vpn_connection_tunnels_up:
@mock_aws
def test_no_vpn_connections(self):
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.vpc.vpc_vpn_connection_tunnels_up.vpc_vpn_connection_tunnels_up.vpc_client",
new=VPC(aws_provider),
):
# Test Check
from prowler.providers.aws.services.vpc.vpc_vpn_connection_tunnels_up.vpc_vpn_connection_tunnels_up import (
vpc_vpn_connection_tunnels_up,
)
check = vpc_vpn_connection_tunnels_up()
result = check.execute()
assert len(result) == 0
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_vpn_both_tunnels_up(self):
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.vpc.vpc_vpn_connection_tunnels_up.vpc_vpn_connection_tunnels_up.vpc_client",
new=VPC(aws_provider),
):
from prowler.providers.aws.services.vpc.vpc_vpn_connection_tunnels_up.vpc_vpn_connection_tunnels_up import (
vpc_vpn_connection_tunnels_up,
)
check = vpc_vpn_connection_tunnels_up()
result = check.execute()
# Se espera que el resultado sea PASS ya que ambos túneles están "UP"
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == "vpn-1234567890abcdef0"
assert (
result[0].resource_arn
== "arn:aws:ec2:us-east-1:123456789012:vpn-connection/vpn-1234567890abcdef0"
)
assert (
result[0].status_extended
== "VPN Connection vpn-1234567890abcdef0 has both tunnels UP. "
)
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call_v2)
def test_vpn_one_tunnel_down(self):
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.vpc.vpc_vpn_connection_tunnels_up.vpc_vpn_connection_tunnels_up.vpc_client",
new=VPC(aws_provider),
):
from prowler.providers.aws.services.vpc.vpc_vpn_connection_tunnels_up.vpc_vpn_connection_tunnels_up import (
vpc_vpn_connection_tunnels_up,
)
check = vpc_vpn_connection_tunnels_up()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == "vpn-1234567890abcdef0"
assert (
result[0].resource_arn
== "arn:aws:ec2:us-east-1:123456789012:vpn-connection/vpn-1234567890abcdef0"
)
assert (
result[0].status_extended
== "VPN Connection vpn-1234567890abcdef0 has at least one tunnel DOWN. "
)