ghndrx/prowler
1
0
Fork 0
mirror of https://github.com/ghndrx/prowler.git synced 2026-02-10 14:55:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(directoryservice): Errors related to the DS Type (#1506)

Browse Source
This commit is contained in:
Pepe Fagoaga
2022-11-21 09:59:37 +01:00
committed by GitHub
parent f99a89eae2
commit 1370e0dec4
14 changed files with 172 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
providers/aws/services/directoryservice/directoryservice_directory_log_forwarding_enabled/directoryservice_directory_log_forwarding_enabled.py
View File

@@ -10,13 +10,13 @@ class directoryservice_directory_log_forwarding_enabled(Check):
for directory in directoryservice_client.directories.values():
report = Check_Report(self.metadata)
report.region = directory.region
report.resource_id = directory.name
report.resource_id = directory.id
if directory.log_subscriptions:
report.status = "PASS"
report.status_extended = f"Directory Service {directory.name} have log forwarding to CloudWatch enabled"
report.status_extended = f"Directory Service {directory.id} have log forwarding to CloudWatch enabled"
else:
report.status = "FAIL"
report.status_extended = f"Directory Service {directory.name} have log forwarding to CloudWatch disabled"
report.status_extended = f"Directory Service {directory.id} have log forwarding to CloudWatch disabled"
findings.append(report)

16
providers/aws/services/directoryservice/directoryservice_directory_log_forwarding_enabled/directoryservice_directory_log_forwarding_enabled_test.py
View File

@@ -3,6 +3,7 @@ from unittest import mock
from providers.aws.services.directoryservice.directoryservice_service import (
Directory,
DirectoryType,
LogSubscriptions,
)
@@ -30,9 +31,12 @@ class Test_directoryservice_directory_log_forwarding_enabled:
def test_one_directory_logging_disabled(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
log_subscriptions=[],
)
@@ -50,20 +54,23 @@ class Test_directoryservice_directory_log_forwarding_enabled:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == "test-directory"
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Directory Service {directory_name} have log forwarding to CloudWatch disabled"
== f"Directory Service {directory_id} have log forwarding to CloudWatch disabled"
)
def test_one_directory_logging_enabled(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
log_subscriptions=[
LogSubscriptions(
@@ -73,6 +80,7 @@ class Test_directoryservice_directory_log_forwarding_enabled:
],
)
}
with mock.patch(
"providers.aws.services.directoryservice.directoryservice_service.DirectoryService",
new=directoryservice_client,
@@ -86,10 +94,10 @@ class Test_directoryservice_directory_log_forwarding_enabled:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == "test-directory"
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Directory Service {directory_name} have log forwarding to CloudWatch enabled"
== f"Directory Service {directory_id} have log forwarding to CloudWatch enabled"
)

6
providers/aws/services/directoryservice/directoryservice_directory_monitor_notifications/directoryservice_directory_monitor_notifications.py
View File

@@ -10,16 +10,16 @@ class directoryservice_directory_monitor_notifications(Check):
for directory in directoryservice_client.directories.values():
report = Check_Report(self.metadata)
report.region = directory.region
report.resource_id = directory.name
report.resource_id = directory.id
if directory.event_topics:
report.status = "PASS"
report.status_extended = (
f"Directory Service {directory.name} have SNS messaging enabled"
f"Directory Service {directory.id} have SNS messaging enabled"
)
else:
report.status = "FAIL"
report.status_extended = (
f"Directory Service {directory.name} have SNS messaging disabled"
f"Directory Service {directory.id} have SNS messaging disabled"
)
findings.append(report)

15
providers/aws/services/directoryservice/directoryservice_directory_monitor_notifications/directoryservice_directory_monitor_notifications_test.py
View File

@@ -5,6 +5,7 @@ from moto.core import DEFAULT_ACCOUNT_ID
from providers.aws.services.directoryservice.directoryservice_service import (
Directory,
DirectoryType,
EventTopics,
EventTopicStatus,
)
@@ -33,8 +34,11 @@ class Test_directoryservice_directory_monitor_notifications:
def test_one_directory_logging_disabled(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
id=directory_id,
type=DirectoryType.MicrosoftAD,
name=directory_name,
region=AWS_REGION,
event_topics=[],
@@ -53,20 +57,23 @@ class Test_directoryservice_directory_monitor_notifications:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == "test-directory"
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Directory Service {directory_name} have SNS messaging disabled"
== f"Directory Service {directory_id} have SNS messaging disabled"
)
def test_one_directory_logging_enabled(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
event_topics=[
EventTopics(
@@ -91,10 +98,10 @@ class Test_directoryservice_directory_monitor_notifications:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == "test-directory"
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Directory Service {directory_name} have SNS messaging enabled"
== f"Directory Service {directory_id} have SNS messaging enabled"
)

8
providers/aws/services/directoryservice/directoryservice_directory_snapshots_limit/directoryservice_directory_snapshots_limit.py
View File

@@ -13,11 +13,11 @@ class directoryservice_directory_snapshots_limit(Check):
for directory in directoryservice_client.directories.values():
report = Check_Report(self.metadata)
report.region = directory.region
report.resource_id = directory.name
report.resource_id = directory.id
if directory.snapshots_limits:
if directory.snapshots_limits.manual_snapshots_limit_reached:
report.status = "FAIL"
report.status_extended = f"Directory Service {directory.name} reached {directory.snapshots_limits.manual_snapshots_limit} Snapshots limit"
report.status_extended = f"Directory Service {directory.id} reached {directory.snapshots_limits.manual_snapshots_limit} Snapshots limit"
else:
limit_remaining = (
directory.snapshots_limits.manual_snapshots_limit
@@ -25,10 +25,10 @@ class directoryservice_directory_snapshots_limit(Check):
)
if limit_remaining <= SNAPSHOT_LIMIT_THRESHOLD:
report.status = "FAIL"
report.status_extended = f"Directory Service {directory.name} is about to reach {directory.snapshots_limits.manual_snapshots_limit} Snapshots which is the limit"
report.status_extended = f"Directory Service {directory.id} is about to reach {directory.snapshots_limits.manual_snapshots_limit} Snapshots which is the limit"
else:
report.status = "PASS"
report.status_extended = f"Directory Service {directory.name} is using {directory.snapshots_limits.manual_snapshots_current_count} out of {directory.snapshots_limits.manual_snapshots_limit} from the Snapshots Limit"
report.status_extended = f"Directory Service {directory.id} is using {directory.snapshots_limits.manual_snapshots_current_count} out of {directory.snapshots_limits.manual_snapshots_limit} from the Snapshots Limit"
findings.append(report)
return findings

29
providers/aws/services/directoryservice/directoryservice_directory_snapshots_limit/directoryservice_directory_snapshots_limit_test.py
View File

@@ -2,6 +2,7 @@ from unittest import mock
from providers.aws.services.directoryservice.directoryservice_service import (
Directory,
DirectoryType,
SnapshotLimit,
)
@@ -29,12 +30,15 @@ class Test_directoryservice_directory_snapshots_limit:
def test_one_directory_snapshots_limit_reached(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
manual_snapshots_current_count = 5
manual_snapshots_limit = 5
manual_snapshots_limit_reached = True
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
snapshots_limits=SnapshotLimit(
manual_snapshots_current_count=manual_snapshots_current_count,
@@ -56,23 +60,26 @@ class Test_directoryservice_directory_snapshots_limit:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == "test-directory"
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Directory Service {directory_name} reached {manual_snapshots_limit} Snapshots limit"
== f"Directory Service {directory_id} reached {manual_snapshots_limit} Snapshots limit"
)
def test_one_directory_snapshots_limit_over_threshold(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
manual_snapshots_current_count = 4
manual_snapshots_limit = 5
manual_snapshots_limit_reached = False
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
snapshots_limits=SnapshotLimit(
manual_snapshots_current_count=manual_snapshots_current_count,
@@ -94,23 +101,26 @@ class Test_directoryservice_directory_snapshots_limit:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == "test-directory"
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Directory Service {directory_name} is about to reach {manual_snapshots_limit} Snapshots which is the limit"
== f"Directory Service {directory_id} is about to reach {manual_snapshots_limit} Snapshots which is the limit"
)
def test_one_directory_snapshots_limit_equal_threshold(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
manual_snapshots_current_count = 3
manual_snapshots_limit = 5
manual_snapshots_limit_reached = False
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
snapshots_limits=SnapshotLimit(
manual_snapshots_current_count=manual_snapshots_current_count,
@@ -132,23 +142,26 @@ class Test_directoryservice_directory_snapshots_limit:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == "test-directory"
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Directory Service {directory_name} is about to reach {manual_snapshots_limit} Snapshots which is the limit"
== f"Directory Service {directory_id} is about to reach {manual_snapshots_limit} Snapshots which is the limit"
)
def test_one_directory_snapshots_limit_more_threshold(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
manual_snapshots_current_count = 1
manual_snapshots_limit = 5
manual_snapshots_limit_reached = False
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
snapshots_limits=SnapshotLimit(
manual_snapshots_current_count=manual_snapshots_current_count,
@@ -170,10 +183,10 @@ class Test_directoryservice_directory_snapshots_limit:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == "test-directory"
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Directory Service {directory_name} is using {manual_snapshots_current_count} out of {manual_snapshots_limit} from the Snapshots Limit"
== f"Directory Service {directory_id} is using {manual_snapshots_current_count} out of {manual_snapshots_limit} from the Snapshots Limit"
)

4
providers/aws/services/directoryservice/directoryservice_ldap_certificate_expiration/directoryservice_ldap_certificate_expiration.py
View File

@@ -23,10 +23,10 @@ class directoryservice_ldap_certificate_expiration(Check):
).days
if remaining_days_to_expire <= DAYS_TO_EXPIRE_THRESHOLD:
report.status = "FAIL"
report.status_extended = f"LDAP Certificate {certificate.id} configured at {directory.name} is about to expire in {remaining_days_to_expire} days"
report.status_extended = f"LDAP Certificate {certificate.id} configured at {directory.id} is about to expire in {remaining_days_to_expire} days"
else:
report.status = "PASS"
report.status_extended = f"LDAP Certificate {certificate.id} configured at {directory.name} expires in {remaining_days_to_expire} days"
report.status_extended = f"LDAP Certificate {certificate.id} configured at {directory.id} expires in {remaining_days_to_expire} days"
findings.append(report)

19
providers/aws/services/directoryservice/directoryservice_ldap_certificate_expiration/directoryservice_ldap_certificate_expiration_test.py
View File

@@ -8,6 +8,7 @@ from providers.aws.services.directoryservice.directoryservice_service import (
CertificateState,
CertificateType,
Directory,
DirectoryType,
)
AWS_REGION = "eu-west-1"
@@ -36,8 +37,11 @@ class Test_directoryservice_ldap_certificate_expiration:
def test_directory_no_certificate(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
id=directory_id,
type=DirectoryType.MicrosoftAD,
name=directory_name,
region=AWS_REGION,
certificates=[],
@@ -63,9 +67,12 @@ class Test_directoryservice_ldap_certificate_expiration:
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
certificate_id = "test-certificate"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
certificates=[
Certificate(
@@ -97,7 +104,7 @@ class Test_directoryservice_ldap_certificate_expiration:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"LDAP Certificate {certificate_id} configured at {directory_name} expires in {remaining_days_to_expire} days"
== f"LDAP Certificate {certificate_id} configured at {directory_id} expires in {remaining_days_to_expire} days"
)
def test_directory_certificate_expires_in_90_days(self):
@@ -106,9 +113,12 @@ class Test_directoryservice_ldap_certificate_expiration:
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
certificate_id = "test-certificate"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
certificates=[
Certificate(
@@ -140,7 +150,7 @@ class Test_directoryservice_ldap_certificate_expiration:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"LDAP Certificate {certificate_id} configured at {directory_name} is about to expire in {remaining_days_to_expire} days"
== f"LDAP Certificate {certificate_id} configured at {directory_id} is about to expire in {remaining_days_to_expire} days"
)
def test_directory_certificate_expires_in_31_days(self):
@@ -149,9 +159,12 @@ class Test_directoryservice_ldap_certificate_expiration:
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
certificate_id = "test-certificate"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
certificates=[
Certificate(
@@ -183,5 +196,5 @@ class Test_directoryservice_ldap_certificate_expiration:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"LDAP Certificate {certificate_id} configured at {directory_name} is about to expire in {remaining_days_to_expire} days"
== f"LDAP Certificate {certificate_id} configured at {directory_id} is about to expire in {remaining_days_to_expire} days"
)

6
providers/aws/services/directoryservice/directoryservice_radius_server_security_protocol/directoryservice_radius_server_security_protocol.py
View File

@@ -14,16 +14,16 @@ class directoryservice_radius_server_security_protocol(Check):
if directory.radius_settings:
report = Check_Report(self.metadata)
report.region = directory.region
report.resource_id = directory.name
report.resource_id = directory.id
if (
directory.radius_settings.authentication_protocol
== AuthenticationProtocol.MS_CHAPv2
):
report.status = "PASS"
report.status_extended = f"Radius server of Directory {directory.name} have recommended security protocol for the Radius server"
report.status_extended = f"Radius server of Directory {directory.id} have recommended security protocol for the Radius server"
else:
report.status = "FAIL"
report.status_extended = f"Radius server of Directory {directory.name} does not have recommended security protocol for the Radius server"
report.status_extended = f"Radius server of Directory {directory.id} does not have recommended security protocol for the Radius server"
findings.append(report)

18
providers/aws/services/directoryservice/directoryservice_radius_server_security_protocol/directoryservice_radius_server_security_protocol_test.py
View File

@@ -3,6 +3,7 @@ from unittest import mock
from providers.aws.services.directoryservice.directoryservice_service import (
AuthenticationProtocol,
Directory,
DirectoryType,
RadiusSettings,
RadiusStatus,
)
@@ -31,9 +32,12 @@ class Test_directoryservice_radius_server_security_protocol:
def test_directory_no_radius_server(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
radius_settings=None,
)
@@ -55,9 +59,12 @@ class Test_directoryservice_radius_server_security_protocol:
def test_directory_radius_server_bad_auth_protocol(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
radius_settings=RadiusSettings(
authentication_protocol=AuthenticationProtocol.MS_CHAPv1,
@@ -78,20 +85,23 @@ class Test_directoryservice_radius_server_security_protocol:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == directory_name
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Radius server of Directory {directory_name} does not have recommended security protocol for the Radius server"
== f"Radius server of Directory {directory_id} does not have recommended security protocol for the Radius server"
)
def test_directory_radius_server_secure_auth_protocol(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
radius_settings=RadiusSettings(
authentication_protocol=AuthenticationProtocol.MS_CHAPv2,
@@ -112,10 +122,10 @@ class Test_directoryservice_radius_server_security_protocol:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == directory_name
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Radius server of Directory {directory_name} have recommended security protocol for the Radius server"
== f"Radius server of Directory {directory_id} have recommended security protocol for the Radius server"
)

37
providers/aws/services/directoryservice/directoryservice_service.py
View File

@@ -44,6 +44,8 @@ class DirectoryService:
for page in describe_fleets_paginator.paginate():
for directory in page["DirectoryDescriptions"]:
directory_id = directory["DirectoryId"]
directory_name = directory["Name"]
directory_type = directory["Type"]
# Radius Configuration
radius_authentication_protocol = (
directory["RadiusSettings"]["AuthenticationProtocol"]
@@ -57,7 +59,9 @@ class DirectoryService:
)
self.directories[directory_id] = Directory(
name=directory_id,
name=directory_name,
id=directory_id,
type=directory_type,
region=regional_client.region,
radius_settings=RadiusSettings(
authentication_protocol=radius_authentication_protocol,
@@ -94,9 +98,7 @@ class DirectoryService:
],
)
)
self.directories[
directory.name
].log_subscriptions = log_subscriptions
self.directories[directory.id].log_subscriptions = log_subscriptions
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -121,7 +123,7 @@ class DirectoryService:
created_date_time=event_topic["CreatedDateTime"],
)
)
self.directories[directory.name].event_topics = event_topics
self.directories[directory.id].event_topics = event_topics
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -131,7 +133,11 @@ class DirectoryService:
logger.info("DirectoryService - Listing Certificates...")
try:
for directory in self.directories.values():
if directory.region == regional_client.region:
# LDAPS operations are not supported for this Directory Type
if (
directory.region == regional_client.region
and directory.type != DirectoryType.SimpleAD
):
list_certificates_paginator = regional_client.get_paginator(
"list_certificates"
)
@@ -150,7 +156,7 @@ class DirectoryService:
type=certificate_info["Type"],
)
)
self.directories[directory.name].certificates = certificates
self.directories[directory.id].certificates = certificates
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -160,12 +166,16 @@ class DirectoryService:
logger.info("DirectoryService - Getting Snapshot Limits...")
try:
for directory in self.directories.values():
if directory.region == regional_client.region:
# Snapshot limits can be fetched only for VPC or Microsoft AD directories.
if (
directory.region == regional_client.region
and directory.type != DirectoryType.ADConnector
):
get_snapshot_limits_parameters = {"DirectoryId": directory.name}
snapshot_limit = regional_client.get_snapshot_limits(
**get_snapshot_limits_parameters
)
self.directories[directory.name].snapshots_limits = SnapshotLimit(
self.directories[directory.id].snapshots_limits = SnapshotLimit(
manual_snapshots_current_count=snapshot_limit["SnapshotLimits"][
"ManualSnapshotsCurrentCount"
],
@@ -250,8 +260,17 @@ class RadiusSettings(BaseModel):
status: Union[RadiusStatus, None]
class DirectoryType(Enum):
SimpleAD = "SimpleAD"
ADConnector = "ADConnector"
MicrosoftAD = "MicrosoftAD"
SharedMicrosoftAD = "SharedMicrosoftAD"
class Directory(BaseModel):
name: str
id: str
type: DirectoryType
log_subscriptions: list[LogSubscriptions] = []
event_topics: list[EventTopics] = []
certificates: list[Certificate] = []

61
providers/aws/services/directoryservice/directoryservice_service_test.py
View File

@@ -11,6 +11,7 @@ from providers.aws.services.directoryservice.directoryservice_service import (
CertificateState,
CertificateType,
DirectoryService,
DirectoryType,
EventTopicStatus,
RadiusStatus,
)
@@ -28,8 +29,9 @@ def mock_make_api_call(self, operation_name, kwarg):
return {
"DirectoryDescriptions": [
{
"DirectoryId": "test-directory",
"DirectoryId": "d-12345a1b2",
"Name": "test-directory",
"Type": "MicrosoftAD",
"ShortName": "test-directory",
"RadiusSettings": {
"RadiusServers": [
@@ -51,7 +53,7 @@ def mock_make_api_call(self, operation_name, kwarg):
return {
"LogSubscriptions": [
{
"DirectoryId": "test-directory",
"DirectoryId": "d-12345a1b2",
"LogGroupName": "test-log-group",
"SubscriptionCreatedDateTime": datetime(2022, 1, 1),
},
@@ -61,7 +63,7 @@ def mock_make_api_call(self, operation_name, kwarg):
return {
"EventTopics": [
{
"DirectoryId": "test-directory",
"DirectoryId": "d-12345a1b2",
"TopicName": "test-topic",
"TopicArn": f"arn:aws:sns:{AWS_REGION}:{DEFAULT_ACCOUNT_ID}:test-topic",
"CreatedDateTime": datetime(2022, 1, 1),
@@ -129,97 +131,100 @@ class Test_DirectoryService_Service:
directoryservice = DirectoryService(current_audit_info)
assert directoryservice.service == "ds"
@mock_ds
def test__describe_directories__(self):
# Set partition for the service
current_audit_info.audited_partition = "aws"
directoryservice = DirectoryService(current_audit_info)
# __describe_directories__
assert directoryservice.directories["test-directory"]
assert directoryservice.directories["test-directory"].name == "test-directory"
assert directoryservice.directories["test-directory"].region == AWS_REGION
assert directoryservice.directories["d-12345a1b2"].id == "d-12345a1b2"
assert (
directoryservice.directories["d-12345a1b2"].type
== DirectoryType.MicrosoftAD
)
assert directoryservice.directories["d-12345a1b2"].name == "test-directory"
assert directoryservice.directories["d-12345a1b2"].region == AWS_REGION
assert (
directoryservice.directories[
"test-directory"
"d-12345a1b2"
].radius_settings.authentication_protocol
== AuthenticationProtocol.MS_CHAPv2
)
assert (
directoryservice.directories["test-directory"].radius_settings.status
directoryservice.directories["d-12345a1b2"].radius_settings.status
== RadiusStatus.Creating
)
# __list_log_subscriptions__
assert len(directoryservice.directories["d-12345a1b2"].log_subscriptions) == 1
assert (
len(directoryservice.directories["test-directory"].log_subscriptions) == 1
)
assert (
directoryservice.directories["test-directory"]
directoryservice.directories["d-12345a1b2"]
.log_subscriptions[0]
.log_group_name
== "test-log-group"
)
assert directoryservice.directories["test-directory"].log_subscriptions[
assert directoryservice.directories["d-12345a1b2"].log_subscriptions[
0
].created_date_time == datetime(2022, 1, 1)
# __describe_event_topics__
assert len(directoryservice.directories["test-directory"].event_topics) == 1
assert len(directoryservice.directories["d-12345a1b2"].event_topics) == 1
assert (
directoryservice.directories["test-directory"].event_topics[0].topic_name
directoryservice.directories["d-12345a1b2"].event_topics[0].topic_name
== "test-topic"
)
assert (
directoryservice.directories["test-directory"].event_topics[0].topic_arn
directoryservice.directories["d-12345a1b2"].event_topics[0].topic_arn
== f"arn:aws:sns:{AWS_REGION}:{DEFAULT_ACCOUNT_ID}:test-topic"
)
assert (
directoryservice.directories["test-directory"].event_topics[0].status
directoryservice.directories["d-12345a1b2"].event_topics[0].status
== EventTopicStatus.Registered
)
assert directoryservice.directories["test-directory"].event_topics[
assert directoryservice.directories["d-12345a1b2"].event_topics[
0
].created_date_time == datetime(2022, 1, 1)
# __list_certificates__
assert len(directoryservice.directories["test-directory"].certificates) == 1
assert len(directoryservice.directories["d-12345a1b2"].certificates) == 1
assert (
directoryservice.directories["test-directory"].certificates[0].id
directoryservice.directories["d-12345a1b2"].certificates[0].id
== "test-certificate"
)
assert (
directoryservice.directories["test-directory"].certificates[0].common_name
directoryservice.directories["d-12345a1b2"].certificates[0].common_name
== "test-certificate"
)
assert (
directoryservice.directories["test-directory"].certificates[0].state
directoryservice.directories["d-12345a1b2"].certificates[0].state
== CertificateState.Registered
)
assert directoryservice.directories["test-directory"].certificates[
assert directoryservice.directories["d-12345a1b2"].certificates[
0
].expiry_date_time == datetime(2023, 1, 1)
assert (
directoryservice.directories["test-directory"].certificates[0].type
directoryservice.directories["d-12345a1b2"].certificates[0].type
== CertificateType.ClientLDAPS
)
# __get_snapshot_limits__
assert directoryservice.directories["test-directory"].snapshots_limits
assert directoryservice.directories["d-12345a1b2"].snapshots_limits
assert (
directoryservice.directories[
"test-directory"
"d-12345a1b2"
].snapshots_limits.manual_snapshots_limit
== 123
)
assert (
directoryservice.directories[
"test-directory"
"d-12345a1b2"
].snapshots_limits.manual_snapshots_current_count
== 123
)
assert (
directoryservice.directories[
"test-directory"
"d-12345a1b2"
].snapshots_limits.manual_snapshots_limit_reached
is True
)

6
providers/aws/services/directoryservice/directoryservice_supported_mfa_radius_enabled/directoryservice_supported_mfa_radius_enabled.py
View File

@@ -14,16 +14,16 @@ class directoryservice_supported_mfa_radius_enabled(Check):
if directory.radius_settings:
report = Check_Report(self.metadata)
report.region = directory.region
report.resource_id = directory.name
report.resource_id = directory.id
if directory.radius_settings.status == RadiusStatus.Completed:
report.status = "PASS"
report.status_extended = (
f"Directory {directory.name} have Radius MFA enabled"
f"Directory {directory.id} have Radius MFA enabled"
)
else:
report.status = "FAIL"
report.status_extended = (
f"Directory {directory.name} does not have Radius MFA enabled"
f"Directory {directory.id} does not have Radius MFA enabled"
)
findings.append(report)

25
providers/aws/services/directoryservice/directoryservice_supported_mfa_radius_enabled/directoryservice_supported_mfa_radius_enabled_test.py
View File

@@ -3,6 +3,7 @@ from unittest import mock
from providers.aws.services.directoryservice.directoryservice_service import (
AuthenticationProtocol,
Directory,
DirectoryType,
RadiusSettings,
RadiusStatus,
)
@@ -31,9 +32,12 @@ class Test_directoryservice_supported_mfa_radius_enabled:
def test_directory_no_radius_server(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
radius_settings=None,
)
@@ -55,9 +59,12 @@ class Test_directoryservice_supported_mfa_radius_enabled:
def test_directory_radius_server_status_failed(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
radius_settings=RadiusSettings(
authentication_protocol=AuthenticationProtocol.MS_CHAPv1,
@@ -78,20 +85,23 @@ class Test_directoryservice_supported_mfa_radius_enabled:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == directory_name
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Directory {directory_name} does not have Radius MFA enabled"
== f"Directory {directory_id} does not have Radius MFA enabled"
)
def test_directory_radius_server_status_creating(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
radius_settings=RadiusSettings(
authentication_protocol=AuthenticationProtocol.MS_CHAPv2,
@@ -112,20 +122,23 @@ class Test_directoryservice_supported_mfa_radius_enabled:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == directory_name
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Directory {directory_name} does not have Radius MFA enabled"
== f"Directory {directory_id} does not have Radius MFA enabled"
)
def test_directory_radius_server_status_completed(self):
directoryservice_client = mock.MagicMock
directory_name = "test-directory"
directory_id = "d-12345a1b2"
directoryservice_client.directories = {
directory_name: Directory(
name=directory_name,
id=directory_id,
type=DirectoryType.MicrosoftAD,
region=AWS_REGION,
radius_settings=RadiusSettings(
authentication_protocol=AuthenticationProtocol.MS_CHAPv2,
@@ -146,10 +159,10 @@ class Test_directoryservice_supported_mfa_radius_enabled:
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == directory_name
assert result[0].resource_id == directory_id
assert result[0].region == AWS_REGION
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Directory {directory_name} have Radius MFA enabled"
== f"Directory {directory_id} have Radius MFA enabled"
)