mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 12:31:54 +00:00
feat(transfer): add new check transfer_server_encryption_in_transit (#5590)
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
committed by
GitHub
parent
ddcd06d9be
commit
5fbf8ddfe9
+32
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"Provider": "aws",
|
||||
"CheckID": "transfer_server_in_transit_encryption_enabled",
|
||||
"CheckTitle": "Transfer Family Servers should have encryption in transit enabled.",
|
||||
"CheckType": [
|
||||
"Software and Configuration Checks/Industry and Regulatory Standards/NIST 800-53 Controls"
|
||||
],
|
||||
"ServiceName": "transfer",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "arn:partition:transfer:region:account-id:server/server-id",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "AwsTransferServer",
|
||||
"Description": "Ensure that your Transfer Family servers have encryption in transit enabled.",
|
||||
"Risk": "Using FTP for endpoint connections leaves data in transit unencrypted, making it susceptible to interception by attackers. FTP lacks encryption, which exposes your data to person-in-the-middle and other interception risks. Adopting encrypted protocols such as SFTP, FTPS, or AS2 provides a layer of protection that helps secure sensitive data during transfer.",
|
||||
"RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/transfer-family-server-no-ftp.html",
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "aws transfer update-server --server-id <server-id> --protocols SFTP FTPS AS2",
|
||||
"NativeIaC": "",
|
||||
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/transfer-controls.html#transfer-2",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Configure AWS Transfer Family servers to use secure protocols, such as SFTP, FTPS, or AS2, instead of FTP to protect data in transit. These protocols offer encryption, reducing exposure to interception and manipulation attacks.",
|
||||
"Url": "https://docs.aws.amazon.com/transfer/latest/userguide/edit-server-config.html#edit-protocols"
|
||||
}
|
||||
},
|
||||
"Categories": [],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+40
@@ -0,0 +1,40 @@
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||
from prowler.providers.aws.services.transfer.transfer_client import transfer_client
|
||||
from prowler.providers.aws.services.transfer.transfer_service import Protocol
|
||||
|
||||
|
||||
class transfer_server_in_transit_encryption_enabled(Check):
|
||||
"""Check if Transfer Servers have encryption in transit enabled.
|
||||
|
||||
This class checks if Transfer Servers have encryption in transit enabled.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[Check_Report_AWS]:
|
||||
"""Execute the server in transit encyption check.
|
||||
|
||||
Iterate over all Transfer Servers and check if they have FTP as one of the valid protocols.
|
||||
|
||||
Returns:
|
||||
List[Check_Report_AWS]: A list of reports for each Transfer Server.
|
||||
"""
|
||||
findings = []
|
||||
for server in transfer_client.servers.values():
|
||||
report = Check_Report_AWS(self.metadata())
|
||||
report.region = server.region
|
||||
report.resource_id = server.id
|
||||
report.resource_arn = server.arn
|
||||
report.resource_tags = server.tags
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"Transfer Server {server.id} does have encryption in transit enabled."
|
||||
)
|
||||
|
||||
if Protocol.FTP in server.protocols:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Transfer Server {server.id} does not have encryption in transit enabled."
|
||||
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -46,8 +46,7 @@ class Transfer(AWSService):
|
||||
)
|
||||
for protocol in server_description.get("Protocols", []):
|
||||
server.protocols.append(Protocol(protocol))
|
||||
for tag in server_description.get("Tags", []):
|
||||
server.tags.append(tag)
|
||||
server.tags = server_description.get("Tags", [])
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{server.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
|
||||
+156
@@ -0,0 +1,156 @@
|
||||
from unittest import mock
|
||||
from unittest.mock import patch
|
||||
|
||||
import botocore
|
||||
from moto import mock_aws
|
||||
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
AWS_REGION_US_EAST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
SERVER_ID = "s-01234567890abcdef"
|
||||
SERVER_ARN = (
|
||||
f"arn:aws:transfer:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:server/{SERVER_ID}"
|
||||
)
|
||||
|
||||
make_api_call = botocore.client.BaseClient._make_api_call
|
||||
|
||||
|
||||
def mock_make_api_call_encrypted(self, operation_name, kwarg):
|
||||
if operation_name == "ListServers":
|
||||
return {
|
||||
"Servers": [
|
||||
{
|
||||
"Arn": f"arn:aws:transfer:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:server/{SERVER_ID}",
|
||||
"ServerId": SERVER_ID,
|
||||
}
|
||||
]
|
||||
}
|
||||
if operation_name == "DescribeServer":
|
||||
return {
|
||||
"Server": {
|
||||
"Arn": SERVER_ARN,
|
||||
"ServerId": SERVER_ID,
|
||||
"Protocols": ["SFTP"],
|
||||
}
|
||||
}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
def mock_make_api_call_unencrypted(self, operation_name, kwarg):
|
||||
if operation_name == "ListServers":
|
||||
return {
|
||||
"Servers": [
|
||||
{
|
||||
"Arn": f"arn:aws:transfer:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:server/{SERVER_ID}",
|
||||
"ServerId": SERVER_ID,
|
||||
}
|
||||
]
|
||||
}
|
||||
if operation_name == "DescribeServer":
|
||||
return {
|
||||
"Server": {
|
||||
"Arn": SERVER_ARN,
|
||||
"ServerId": SERVER_ID,
|
||||
"Protocols": ["FTP", "FTPS", "SFTP", "AS2"],
|
||||
}
|
||||
}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
class Test_transfer_server_encryption_in_transit:
|
||||
@mock_aws
|
||||
def test_no_servers(self):
|
||||
from prowler.providers.aws.services.transfer.transfer_service import Transfer
|
||||
|
||||
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,
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.transfer.transfer_server_in_transit_encryption_enabled.transfer_server_in_transit_encryption_enabled.transfer_client",
|
||||
new=Transfer(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.transfer.transfer_server_in_transit_encryption_enabled.transfer_server_in_transit_encryption_enabled import (
|
||||
transfer_server_in_transit_encryption_enabled,
|
||||
)
|
||||
|
||||
check = transfer_server_in_transit_encryption_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call", new=mock_make_api_call_encrypted
|
||||
)
|
||||
@mock_aws
|
||||
def test_transfer_server_encryption_enabled(self):
|
||||
from prowler.providers.aws.services.transfer.transfer_service import Transfer
|
||||
|
||||
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,
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.transfer.transfer_server_in_transit_encryption_enabled.transfer_server_in_transit_encryption_enabled.transfer_client",
|
||||
new=Transfer(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.transfer.transfer_server_in_transit_encryption_enabled.transfer_server_in_transit_encryption_enabled import (
|
||||
transfer_server_in_transit_encryption_enabled,
|
||||
)
|
||||
|
||||
check = transfer_server_in_transit_encryption_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"Transfer Server {SERVER_ID} does have encryption in transit enabled."
|
||||
)
|
||||
assert result[0].resource_id == SERVER_ID
|
||||
assert result[0].resource_arn == SERVER_ARN
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call", new=mock_make_api_call_unencrypted
|
||||
)
|
||||
@mock_aws
|
||||
def test_transfer_server_encryption_disabled(self):
|
||||
from prowler.providers.aws.services.transfer.transfer_service import Transfer
|
||||
|
||||
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,
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.transfer.transfer_server_in_transit_encryption_enabled.transfer_server_in_transit_encryption_enabled.transfer_client",
|
||||
new=Transfer(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.transfer.transfer_server_in_transit_encryption_enabled.transfer_server_in_transit_encryption_enabled import (
|
||||
transfer_server_in_transit_encryption_enabled,
|
||||
)
|
||||
|
||||
check = transfer_server_in_transit_encryption_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"Transfer Server {SERVER_ID} does not have encryption in transit enabled."
|
||||
)
|
||||
assert result[0].resource_id == SERVER_ID
|
||||
assert result[0].resource_arn == SERVER_ARN
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
@@ -16,6 +16,9 @@ SERVER_ID = "SERVICE_MANAGED::s-01234567890abcdef"
|
||||
SERVER_ARN = f"arn:aws:transfer:us-east-1:{AWS_ACCOUNT_NUMBER}:server/{SERVER_ID}"
|
||||
|
||||
|
||||
make_api_call = botocore.client.BaseClient._make_api_call
|
||||
|
||||
|
||||
def mock_make_api_call(self, operation_name, kwarg):
|
||||
if operation_name == "ListServers":
|
||||
return {
|
||||
@@ -32,7 +35,7 @@ def mock_make_api_call(self, operation_name, kwarg):
|
||||
"Arn": SERVER_ARN,
|
||||
"ServerId": SERVER_ID,
|
||||
"Protocols": ["SFTP"],
|
||||
"Tags": [{"key": "value"}],
|
||||
"Tags": [{"Key": "key", "Value": "value"}],
|
||||
}
|
||||
}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
@@ -78,5 +81,5 @@ class Test_transfer_service:
|
||||
assert transfer.servers[SERVER_ARN].id == SERVER_ID
|
||||
assert len(transfer.servers[SERVER_ARN].protocols) == 1
|
||||
assert transfer.servers[SERVER_ARN].region == "us-east-1"
|
||||
assert transfer.servers[SERVER_ARN].tags == [{"key": "value"}]
|
||||
assert transfer.servers[SERVER_ARN].tags == [{"Key": "key", "Value": "value"}]
|
||||
assert transfer.servers[SERVER_ARN].protocols[0] == Protocol.SFTP
|
||||
|
||||
Reference in New Issue
Block a user