mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
feat(config): add new check config_recorder_using_aws_service_role (#5357)
This commit is contained in:
Rubén De la Torre Vico
committed by
GitHub
GitHub
parent
e0ed891fc4
commit
037e40f8e4
+2
-4
@@ -5,15 +5,13 @@ from prowler.providers.aws.services.config.config_client import config_client
|
||||
class config_recorder_all_regions_enabled(Check):
|
||||
def execute(self):
|
||||
findings = []
|
||||
for recorder in config_client.recorders:
|
||||
for recorder in config_client.recorders.values():
|
||||
report = Check_Report_AWS(self.metadata())
|
||||
report.region = recorder.region
|
||||
report.resource_arn = config_client._get_recorder_arn_template(
|
||||
recorder.region
|
||||
)
|
||||
report.resource_id = (
|
||||
config_client.audited_account if not recorder.name else recorder.name
|
||||
)
|
||||
report.resource_id = recorder.name
|
||||
# Check if Config is enabled in region
|
||||
if not recorder.name:
|
||||
report.status = "FAIL"
|
||||
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"Provider": "aws",
|
||||
"CheckID": "config_recorder_using_aws_service_role",
|
||||
"CheckTitle": "Ensure Config Recorder is using service-linked AWS Config role",
|
||||
"CheckType": [
|
||||
"Software and Configuration Checks/AWS Security Best Practices/AWS Foundational Security Best Practices"
|
||||
],
|
||||
"ServiceName": "config",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "arn:partition:access-recorder:region:account-id:recorder/resource-id",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "Other",
|
||||
"Description": "AWS Config uses an IAM role to access other AWS services. This role should be AWSServiceRoleForConfig, not a custom role. Using AWSServiceRoleForConfig ensures that the Config recorder has the necessary permissions to record configuration changes and that the role is managed by AWS, reducing the risk of misconfiguration.",
|
||||
"Risk": "If the Config recorder is not using AWSServiceRoleForConfig, it may not have the necessary permissions to record configuration changes, which could lead in not following the principle of least privilege, which could lead to misconfiguration and potential security vulnerabilities.",
|
||||
"RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/using-service-linked-roles.html",
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "aws configservice put-configuration-recorder --configuration-recorder- name=<recorder-name>,roleARN=arn:<audited_partition>:iam::<account_number>:role/aws-service-role/config.amazonaws.com/AWSServiceRoleForConfig",
|
||||
"NativeIaC": "",
|
||||
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/config-controls.html#config-1",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Use service-linked role AWSServiceRoleForConfig for AWS Config recorders.",
|
||||
"Url": "https://docs.aws.amazon.com/config/latest/developerguide/using-service-linked-roles.html"
|
||||
}
|
||||
},
|
||||
"Categories": [],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+28
@@ -0,0 +1,28 @@
|
||||
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||
from prowler.providers.aws.services.config.config_client import config_client
|
||||
|
||||
|
||||
class config_recorder_using_aws_service_role(Check):
|
||||
def execute(self):
|
||||
findings = []
|
||||
for recorder in config_client.recorders.values():
|
||||
if recorder.name and recorder.recording:
|
||||
report = Check_Report_AWS(self.metadata())
|
||||
report.region = recorder.region
|
||||
report.resource_arn = config_client._get_recorder_arn_template(
|
||||
recorder.region
|
||||
)
|
||||
report.resource_id = recorder.name
|
||||
if (
|
||||
recorder.role_arn
|
||||
== f"arn:{config_client.audited_partition}:iam::{config_client.audited_account}:role/aws-service-role/config.amazonaws.com/AWSServiceRoleForConfig"
|
||||
):
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"AWS Config recorder {recorder.name} is using AWSServiceRoleForConfig."
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"AWS Config recorder {recorder.name} is not using AWSServiceRoleForConfig."
|
||||
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -7,66 +7,76 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered
|
||||
from prowler.providers.aws.lib.service.service import AWSService
|
||||
|
||||
|
||||
################## Config
|
||||
class Config(AWSService):
|
||||
def __init__(self, provider):
|
||||
# Call AWSService's __init__
|
||||
super().__init__(__class__.__name__, provider)
|
||||
self.recorders = []
|
||||
self.__threading_call__(self._describe_configuration_recorder_status)
|
||||
self.recorders = {}
|
||||
self.__threading_call__(self.describe_configuration_recorders)
|
||||
self.__threading_call__(
|
||||
self._describe_configuration_recorder_status, self.recorders.values()
|
||||
)
|
||||
|
||||
def _get_recorder_arn_template(self, region):
|
||||
return f"arn:{self.audited_partition}:config:{region}:{self.audited_account}:recorder"
|
||||
|
||||
def _describe_configuration_recorder_status(self, regional_client):
|
||||
def describe_configuration_recorders(self, regional_client):
|
||||
logger.info("Config - Listing Recorders...")
|
||||
try:
|
||||
recorders = regional_client.describe_configuration_recorder_status()[
|
||||
"ConfigurationRecordersStatus"
|
||||
]
|
||||
recorders_count = 0
|
||||
for recorder in recorders:
|
||||
if not self.audit_resources or (
|
||||
is_resource_filtered(recorder["name"], self.audit_resources)
|
||||
):
|
||||
recorders_count += 1
|
||||
if "lastStatus" in recorder:
|
||||
self.recorders.append(
|
||||
Recorder(
|
||||
name=recorder["name"],
|
||||
recording=recorder["recording"],
|
||||
last_status=recorder["lastStatus"],
|
||||
region=regional_client.region,
|
||||
)
|
||||
)
|
||||
else:
|
||||
self.recorders.append(
|
||||
Recorder(
|
||||
name=recorder["name"],
|
||||
recording=recorder["recording"],
|
||||
last_status=None,
|
||||
region=regional_client.region,
|
||||
)
|
||||
)
|
||||
recorders = regional_client.describe_configuration_recorders().get(
|
||||
"ConfigurationRecorders", []
|
||||
)
|
||||
|
||||
# No config recorders in region
|
||||
if recorders_count == 0:
|
||||
self.recorders.append(
|
||||
Recorder(
|
||||
name=self.audited_account,
|
||||
recording=None,
|
||||
last_status=None,
|
||||
region=regional_client.region,
|
||||
)
|
||||
if not recorders:
|
||||
self.recorders[regional_client.region] = Recorder(
|
||||
name=self.audited_account,
|
||||
role_arn="",
|
||||
region=regional_client.region,
|
||||
)
|
||||
else:
|
||||
for recorder in recorders:
|
||||
if not self.audit_resources or (
|
||||
is_resource_filtered(recorder["name"], self.audit_resources)
|
||||
):
|
||||
self.recorders[recorder["name"]] = Recorder(
|
||||
name=recorder["name"],
|
||||
role_arn=recorder["roleARN"],
|
||||
region=regional_client.region,
|
||||
)
|
||||
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _describe_configuration_recorder_status(self, recorder):
|
||||
logger.info("Config - Listing Recorders Status...")
|
||||
try:
|
||||
if recorder.name != self.audited_account:
|
||||
recorder_status = (
|
||||
self.regional_clients[recorder.region]
|
||||
.describe_configuration_recorder_status(
|
||||
ConfigurationRecorderNames=[recorder.name]
|
||||
)
|
||||
.get("ConfigurationRecordersStatus", [])
|
||||
)
|
||||
|
||||
if recorder_status:
|
||||
recorder.recording = recorder_status[0].get("recording", False)
|
||||
recorder.last_status = recorder_status[0].get(
|
||||
"lastStatus", "Failure"
|
||||
)
|
||||
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{recorder.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
|
||||
class Recorder(BaseModel):
|
||||
name: str
|
||||
role_arn: str
|
||||
recording: Optional[bool]
|
||||
last_status: Optional[str]
|
||||
region: str
|
||||
|
||||
+163
@@ -0,0 +1,163 @@
|
||||
from unittest import mock
|
||||
|
||||
from boto3 import client
|
||||
from moto import mock_aws
|
||||
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
AWS_REGION_EU_WEST_1,
|
||||
AWS_REGION_US_EAST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_config_recorder_using_aws_service_role:
|
||||
@mock_aws
|
||||
def test_config_no_recorders(self):
|
||||
from prowler.providers.aws.services.config.config_service import Config
|
||||
|
||||
aws_provider = set_mocked_aws_provider(
|
||||
[AWS_REGION_EU_WEST_1, 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.config.config_recorder_using_aws_service_role.config_recorder_using_aws_service_role.config_client",
|
||||
new=Config(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.config.config_recorder_using_aws_service_role.config_recorder_using_aws_service_role import (
|
||||
config_recorder_using_aws_service_role,
|
||||
)
|
||||
|
||||
check = config_recorder_using_aws_service_role()
|
||||
results = check.execute()
|
||||
|
||||
assert len(results) == 0
|
||||
|
||||
@mock_aws
|
||||
def test_config_one_recoder_disabled(self):
|
||||
# Create Config Mocked Resources
|
||||
config_client = client("config", region_name=AWS_REGION_US_EAST_1)
|
||||
# Create Config Recorder
|
||||
config_client.put_configuration_recorder(
|
||||
ConfigurationRecorder={"name": "default", "roleARN": "somearn"}
|
||||
)
|
||||
from prowler.providers.aws.services.config.config_service import Config
|
||||
|
||||
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.config.config_recorder_using_aws_service_role.config_recorder_using_aws_service_role.config_client",
|
||||
new=Config(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.config.config_recorder_using_aws_service_role.config_recorder_using_aws_service_role import (
|
||||
config_recorder_using_aws_service_role,
|
||||
)
|
||||
|
||||
check = config_recorder_using_aws_service_role()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
@mock_aws
|
||||
def test_config_recorder_using_aws_service_role(self):
|
||||
# Create Config Mocked Resources
|
||||
config_client = client("config", region_name=AWS_REGION_US_EAST_1)
|
||||
# Create Config Recorder
|
||||
config_client.put_configuration_recorder(
|
||||
ConfigurationRecorder={
|
||||
"name": "default",
|
||||
"roleARN": "arn:aws:iam::123456789012:role/aws-service-role/config.amazonaws.com/AWSServiceRoleForConfig",
|
||||
}
|
||||
)
|
||||
# Make the delivery channel
|
||||
config_client.put_delivery_channel(
|
||||
DeliveryChannel={"name": "testchannel", "s3BucketName": "somebucket"}
|
||||
)
|
||||
config_client.start_configuration_recorder(ConfigurationRecorderName="default")
|
||||
from prowler.providers.aws.services.config.config_service import Config
|
||||
|
||||
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.config.config_recorder_using_aws_service_role.config_recorder_using_aws_service_role.config_client",
|
||||
new=Config(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.config.config_recorder_using_aws_service_role.config_recorder_using_aws_service_role import (
|
||||
config_recorder_using_aws_service_role,
|
||||
)
|
||||
|
||||
check = config_recorder_using_aws_service_role()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
# Search for the recorder just created
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "AWS Config recorder default is using AWSServiceRoleForConfig."
|
||||
)
|
||||
assert result[0].resource_id == "default"
|
||||
assert (
|
||||
result[0].resource_arn
|
||||
== f"arn:aws:config:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:recorder"
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_tags == []
|
||||
|
||||
@mock_aws
|
||||
def test_config_recorder_not_using_aws_service_role(self):
|
||||
# Create Config Mocked Resources
|
||||
config_client = client("config", region_name=AWS_REGION_US_EAST_1)
|
||||
# Create Config Recorder
|
||||
config_client.put_configuration_recorder(
|
||||
ConfigurationRecorder={
|
||||
"name": "default",
|
||||
"roleARN": "arn:aws:iam::123456789012:role/MyCustomRole",
|
||||
}
|
||||
)
|
||||
# Make the delivery channel
|
||||
config_client.put_delivery_channel(
|
||||
DeliveryChannel={"name": "testchannel", "s3BucketName": "somebucket"}
|
||||
)
|
||||
config_client.start_configuration_recorder(ConfigurationRecorderName="default")
|
||||
from prowler.providers.aws.services.config.config_service import Config
|
||||
|
||||
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.config.config_recorder_using_aws_service_role.config_recorder_using_aws_service_role.config_client",
|
||||
new=Config(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.config.config_recorder_using_aws_service_role.config_recorder_using_aws_service_role import (
|
||||
config_recorder_using_aws_service_role,
|
||||
)
|
||||
|
||||
check = config_recorder_using_aws_service_role()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== "AWS Config recorder default is not using AWSServiceRoleForConfig."
|
||||
)
|
||||
assert result[0].resource_id == "default"
|
||||
assert (
|
||||
result[0].resource_arn
|
||||
== f"arn:aws:config:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:recorder"
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_tags == []
|
||||
@@ -52,6 +52,30 @@ class Test_Config_Service:
|
||||
config = Config(aws_provider)
|
||||
assert config.audited_account == AWS_ACCOUNT_NUMBER
|
||||
|
||||
@mock_aws
|
||||
def test_describe_configuration_recorders(self):
|
||||
# Generate Config Client
|
||||
config_client = client("config", region_name=AWS_REGION_EU_WEST_1)
|
||||
# Create Config Recorder and start it
|
||||
config_client.put_configuration_recorder(
|
||||
ConfigurationRecorder={"name": "default", "roleARN": "somearn"}
|
||||
)
|
||||
# Make the delivery channel
|
||||
config_client.put_delivery_channel(
|
||||
DeliveryChannel={"name": "testchannel", "s3BucketName": "somebucket"}
|
||||
)
|
||||
config_client.start_configuration_recorder(ConfigurationRecorderName="default")
|
||||
# Config client for this test class
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
|
||||
config = Config(aws_provider)
|
||||
# One recorder per region
|
||||
assert len(config.recorders) == 1
|
||||
# Check the active one
|
||||
assert "default" in config.recorders
|
||||
assert config.recorders["default"].name == "default"
|
||||
assert config.recorders["default"].role_arn == "somearn"
|
||||
assert config.recorders["default"].region == AWS_REGION_EU_WEST_1
|
||||
|
||||
# Test Config Get Rest APIs
|
||||
@mock_aws
|
||||
def test_describe_configuration_recorder_status(self):
|
||||
@@ -75,6 +99,6 @@ class Test_Config_Service:
|
||||
assert len(config.recorders) == 2
|
||||
# Check the active one
|
||||
# Search for the recorder just created
|
||||
for recorder in config.recorders:
|
||||
for recorder in config.recorders.values():
|
||||
if recorder.name == "default":
|
||||
assert recorder.recording is True
|
||||
|
||||
Reference in New Issue
Block a user